Python FastAPI E-Learning Part 5: Pipeline Data Excel Berkecepatan Tinggi & Pengujian Asinkron Otomatis dengan Pytest

# backend/app/services/excel_service.py
import io
import openpyxl
from typing import List, Dict, Any

class ExcelService:
    @staticmethod
    def parse_student_excel(file_bytes: bytes) -> List[Dict[str, Any]]:
        # Muat workbook langsung dari stream memori tanpa menyentuh disk fisik
        wb = openpyxl.load_workbook(filename=io.BytesIO(file_bytes), data_only=True)
        sheet = wb.active

        students = []
        # Header baris 1 yang diharapkan: [Nama, Email, NIS, NISN, Gender, Telepon, Alamat]
        # iter_rows(values_only=True) melewati objek pembungkus Cell berat secara efisien
        for row_idx, row in enumerate(sheet.iter_rows(values_only=True), start=1):
            if row_idx == 1:
                continue  # Lewati baris header judul kolom
            if not row or not any(row):
                continue  # Lewati baris kosong

            name = str(row[0]).strip() if len(row) > 0 and row[0] is not None else None
            email = str(row[1]).strip() if len(row) > 1 and row[1] is not None else None
            nis = str(row[2]).strip() if len(row) > 2 and row[2] is not None else None
            nisn = str(row[3]).strip() if len(row) > 3 and row[3] is not None else None
            gender = str(row[4]).strip().upper() if len(row) > 4 and row[4] is not None else "L"
            phone = str(row[5]).strip() if len(row) > 5 and row[5] is not None else None
            address = str(row[6]).strip() if len(row) > 6 and row[6] is not None else None

            # Validasi kolom identitas wajib
            if name and email and nis:
                students.append({
                    "name": name,
                    "email": email,
                    "nis": nis,
                    "nisn": nisn,
                    "gender": gender if gender in ["L", "P", "M", "F"] else "L",
                    "phone": phone,
                    "address": address,
                })

        return students

    @staticmethod
    def parse_questions_excel(file_bytes: bytes) -> List[Dict[str, Any]]:
        wb = openpyxl.load_workbook(filename=io.BytesIO(file_bytes), data_only=True)
        sheet = wb.active

        questions = []
        # Header: [Teks Soal, Tipe, Opsi A, Opsi B, Opsi C, Opsi D, Kunci Jawaban, Bobot, Kesulitan]
        for row_idx, row in enumerate(sheet.iter_rows(values_only=True), start=1):
            if row_idx == 1:
                continue
            if not row or not any(row):
                continue

            q_text = str(row[0]).strip() if len(row) > 0 and row[0] is not None else None
            q_type = str(row[1]).strip().lower() if len(row) > 1 and row[1] is not None else "multiple_choice"
            opt_a = str(row[2]).strip() if len(row) > 2 and row[2] is not None else None
            opt_b = str(row[3]).strip() if len(row) > 3 and row[3] is not None else None
            opt_c = str(row[4]).strip() if len(row) > 4 and row[4] is not None else None
            opt_d = str(row[5]).strip() if len(row) > 5 and row[5] is not None else None
            correct = str(row[6]).strip().upper() if len(row) > 6 and row[6] is not None else "A"
            points = int(row[7]) if len(row) > 7 and row[7] is not None and str(row[7]).isdigit() else 1
            difficulty = str(row[8]).strip().lower() if len(row) > 8 and row[8] is not None else "medium"

            if q_text:
                options = []
                if opt_a: options.append({"key": "A", "text": opt_a})
                if opt_b: options.append({"key": "B", "text": opt_b})
                if opt_c: options.append({"key": "C", "text": opt_c})
                if opt_d: options.append({"key": "D", "text": opt_d})

                questions.append({
                    "question_text": q_text,
                    "question_type": q_type,
                    "options": options if options else None,
                    "correct_answer": correct,
                    "points": points,
                    "difficulty": difficulty if difficulty in ["easy", "medium", "hard"] else "medium",
                })

        return questions

excel_service = ExcelService()

# backend/app/api/v1/students.py
@router.post("/import-excel", response_model=ApiResponse[Dict[str, int]])
async def import_students(
    classroom_id: int,
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db),
    admin: User = Depends(require_roles(["ROLE_ADMIN"]))
):
    content = await file.read()
    records = excel_service.parse_student_excel(content)
    if not records:
        raise HTTPException(status_code=400, detail="Berkas Excel tidak memuat data yang valid")

    # Ambil role siswa standar
    role_stmt = select(Role).where(Role.name == "ROLE_STUDENT")
    student_role = (await db.execute(role_stmt)).scalars().first()

    default_hashed_pwd = get_password_hash("password")
    imported_count = 0

    # Blok transaksi atomik
    try:
        for item in records:
            # 1. Buat Akun Pengguna Login
            user = User(
                name=item["name"],
                email=item["email"],
                password=default_hashed_pwd,
                is_active=True,
                roles=[student_role]
            )
            db.add(user)
            await db.flush()  # Flush ke DB untuk memperoleh user.id tanpa commit permanen

            # 2. Buat Profil Siswa dengan Foreign Key user_id
            student = Student(
                user_id=user.id,
                classroom_id=classroom_id,
                nis=item["nis"],
                nisn=item.get("nisn"),
                gender=item["gender"],
                phone=item.get("phone"),
                address=item.get("address"),
            )
            db.add(student)
            imported_count += 1

        # Commit seluruh 2.000 data sekaligus secara serentak
        await db.commit()

    except Exception as exc:
        await db.rollback()
        raise HTTPException(
            status_code=400,
            detail=f"Impor massal dibatalkan karena pelanggaran batasan data: {str(exc)}"
        )

    return ApiResponse.ok(
        data={"imported": imported_count},
        message=f"Berhasil mengimpor {imported_count} siswa ke dalam kelas"
    )

# backend/tests/test_excel_import.py
import io
import openpyxl
from app.services.excel_service import excel_service

def test_excel_student_parsing():
    # 1. Bangun spreadsheet sintetis di memori
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.append(["Nama", "Email", "NIS", "NISN", "Gender", "Telepon", "Alamat"])
    ws.append(["Ahmad Test", "[email protected]", "20249999", "00999999", "L", "0811111111", "Jakarta"])
    
    stream = io.BytesIO()
    wb.save(stream)
    stream.seek(0)
    
    # 2. Eksekusi parser
    students = excel_service.parse_student_excel(stream.getvalue())
    
    # 3. Verifikasi struktur data yang dihasilkan
    assert len(students) == 1
    assert students[0]["name"] == "Ahmad Test"
    assert students[0]["nis"] == "20249999"
    assert students[0]["gender"] == "L"

def test_excel_question_parsing():
    wb = openpyxl.Workbook()
    ws = wb.active
    ws.append(["Question Text", "Type", "Option A", "Option B", "Option C", "Option D", "Correct Answer", "Points", "Difficulty"])
    ws.append(["Apa ibukota Indonesia?", "multiple_choice", "Bandung", "Jakarta", "Surabaya", "Medan", "B", 10, "easy"])
    
    stream = io.BytesIO()
    wb.save(stream)
    stream.seek(0)
    
    questions = excel_service.parse_questions_excel(stream.getvalue())
    assert len(questions) == 1
    assert questions[0]["question_text"] == "Apa ibukota Indonesia?"
    assert questions[0]["correct_answer"] == "B"
    assert questions[0]["points"] == 10
    assert len(questions[0]["options"]) == 4

# backend/tests/test_auth.py
import pytest
from httpx import AsyncClient
from app.main import app

@pytest.mark.asyncio
async def test_login_success():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post(
            "/api/v1/auth/login",
            json={"email": "[email protected]", "password": "password"}
        )
        assert response.status_code == 200
        json_data = response.json()
        assert json_data["success"] is True
        assert "token" in json_data["data"]
        assert json_data["data"]["user"]["email"] == "[email protected]"

@pytest.mark.asyncio
async def test_unauthorized_access():
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Panggil endpoint terproteksi tanpa menyertakan bearer token
        response = await client.get("/api/v1/admin/dashboard")
        assert response.status_code == 401

# .github/workflows/backend-ci.yml
name: Backend CI

on:
  push:
    branches: [ main ]
  pull_request:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: elearning_test
          POSTGRES_USER: postgres
          POSTGRES_PASSWORD: postgres
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python 3.12
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"
          cache: "pip"

      - name: Install Dependencies
        working-directory: backend
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-asyncio httpx coverage

      - name: Eksekusi Suite Pytest dengan Coverage
        working-directory: backend
        env:
          DATABASE_URL: postgresql+asyncpg://postgres:postgres@localhost:5432/elearning_test
          REDIS_URL: redis://localhost:6379/0
          SECRET_KEY: test_secret_key_ci_cd
        run: |
          pytest -v --cov=app --cov-report=term-missing