The Data Pipeline & Quality Assurance Challenge
In educational systems, administrative workflows frequently require bulk data ingestion:
- Academic Enrollment: Importing spreadsheets containing 2,000 newly enrolled students across 10 classrooms at the beginning of an academic year.
- CBT Question Banks: Teachers uploading standardized test spreadsheets with 200 multiple-choice questions, answer keys, points, and explanations.
Naively reading large .xlsx files using conventional DOM-based XML parsers allocates heavy memory footprints (often 50MB–100MB per file), leading to out-of-memory (OOM) worker restarts under concurrent uploads. Furthermore, if row #1,850 contains an invalid email address, a naive loop leaves 1,849 orphaned user accounts in the database without corresponding student profiles.
In mfarim/fastapi-elearning-react, we engineered:
- Memory-Safe Excel Streaming Ingestion: Using
openpyxl'siter_rows(values_only=True). - Atomic Batch Database Transactions: Guaranteeing all-or-nothing database integrity.
- Async Automated Testing Suite: With
pytest,pytest-asyncio, andhttpx. - Automated GitHub Actions CI/CD: Running automated tests against ephemeral PostgreSQL and Redis service containers.
┌────────────────────────────────────────────────────────────────────────────────────────┐
│ BULK DATA INGESTION & PIPELINE TOPOLOGY │
└────────────────────────────────────────────────────────────────────────────────────────┘
📥 Teacher / Admin uploads .xlsx spreadsheet (e.g. 2,000 rows)
│
▼ HTTP Multipart Upload (io.BytesIO)
┌──────────────────────────────────────────────────────────────────────────────────┐
│ FASTAPI EXCEL SERVICE: Streaming Memory Reader (openpyxl) │
│ • iter_rows(values_only=True) reads raw data tuples │
│ • Zero XML DOM cell object allocation (RAM stays < 15MB) │
│ • Data cleaning, whitespace trimming, uppercase normalization │
└───────────────────────┬──────────────────────────────────────────────────────────┘
│
▼ Validated Record Dictionaries
┌──────────────────────────────────────────────────────────────────────────────────┐
│ ASYNC DATABASE INGESTION (Atomic Transaction Scope): │
│ • Generates BCrypt password hashes in parallel │
│ • Bulk inserts into 'users' table │
│ • Bulk inserts into 'students' table with foreign key linkage │
│ • Automatic ROLLBACK if any row triggers database constraint violation │
└───────────────────────┬──────────────────────────────────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────────────────────────────┐
│ PYTEST AUTOMATION & CI/CD PIPELINE: │
│ • Ephemeral PostgreSQL 16 & Redis 7 Docker services │
│ • httpx.AsyncClient dispatches async HTTP calls directly to ASGI app │
│ • 100% test pass required before merging to main branch │
└──────────────────────────────────────────────────────────────────────────────────┘
1. High-Throughput Streaming Excel Reader with openpyxl
In backend/app/services/excel_service.py, our spreadsheet reader is engineered for high throughput and minimal memory consumption:
PYTHON
# 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]]:
# Load workbook in memory without saving to physical disk
wb = openpyxl.load_workbook(filename=io.BytesIO(file_bytes), data_only=True)
sheet = wb.active
students = []
# Expected header row 1: [Name, Email, Student ID (NIS), NISN, Gender, Phone, Address]
# iter_rows(values_only=True) skips heavy Cell wrapper objects entirely
for row_idx, row in enumerate(sheet.iter_rows(values_only=True), start=1):
if row_idx == 1:
continue # Skip header row
if not row or not any(row):
continue # Skip blank rows
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
# Validate mandatory primary identity fields
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 = []
# Expected header: [Question Text, Type, Option A, Option B, Option C, Option D, Correct Answer, Points, Difficulty]
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()
Why iter_rows(values_only=True) is Critical
A standard openpyxl call like sheet['A1'] wraps every cell in an elaborate Python Cell object that tracks fonts, borders, background colors, and formula trees. In a sheet with 2,000 rows and 10 columns (20,000 cells), allocating these objects consumes over 60MB of RAM.
Using iter_rows(values_only=True) instructs the underlying XML parser to extract raw Python primitive strings and integers, dropping memory consumption to less than 12MB for the entire parse operation.
2. Atomic Batch Database Ingestion
In backend/app/api/v1/students.py, importing students requires creating both a User record (for authentication) and a Student record (with classroom association). This must happen inside an atomic database transaction:
PYTHON
# 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="Spreadsheet contains no valid records")
# Fetch default student role
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
# Atomic transaction block
try:
for item in records:
# 1. Provision User Account
user = User(
name=item["name"],
email=item["email"],
password=default_hashed_pwd,
is_active=True,
roles=[student_role]
)
db.add(user)
await db.flush() # Flushes to DB to obtain user.id without committing
# 2. Provision Student Profile with Foreign Key
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 all 2,000 records simultaneously
await db.commit()
except Exception as exc:
await db.rollback()
raise HTTPException(
status_code=400,
detail=f"Batch import aborted due to database constraint violation: {str(exc)}"
)
return ApiResponse.ok(
data={"imported": imported_count},
message=f"Successfully imported {imported_count} students into classroom"
)
By calling await db.flush() inside the loop, the database assigns primary keys (user.id) needed for foreign key relationships without committing the transaction. If any duplicate email or NIS violation occurs on row #1,999, the except block executes await db.rollback(), ensuring zero partial data corruption.
3. Automated Async Testing with Pytest & Pytest-Asyncio
In asynchronous Python applications, traditional synchronous test runners (unittest.TestCase) cannot await coroutines. We use Pytest with pytest-asyncio and httpx.AsyncClient.
Synthetic In-Memory Excel Unit Testing (test_excel_import.py)
In backend/tests/test_excel_import.py, we test Excel parsing without reading files from disk by generating synthetic binary streams in memory:
PYTHON
# backend/tests/test_excel_import.py
import io
import openpyxl
from app.services.excel_service import excel_service
def test_excel_student_parsing():
# 1. Create synthetic in-memory workbook
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. Execute parser
students = excel_service.parse_student_excel(stream.getvalue())
# 3. Assert correct structure
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(["What is the capital of 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"] == "What is the capital of Indonesia?"
assert questions[0]["correct_answer"] == "B"
assert questions[0]["points"] == 10
assert len(questions[0]["options"]) == 4
End-to-End Async Route Testing (test_auth.py)
In backend/tests/test_auth.py, we test authentication routes asynchronously using httpx.AsyncClient:
PYTHON
# 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:
# Access protected route without token
response = await client.get("/api/v1/admin/dashboard")
assert response.status_code == 401
4. Continuous Integration via GitHub Actions
In .github/workflows/backend-ci.yml, every push and pull request triggers an automated test run against real, isolated database services:
YAML
# .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: Set up 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: Run Pytest Suite with 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
Key Takeaways for Engineering Leads
- Always use streaming readers for bulk file uploads:
openpyxl.load_workbook(data_only=True)combined withsheet.iter_rows(values_only=True)consumes 80% less memory than standard cell access. - Never commit inside a batch loop: Use
await db.flush()to obtain auto-generated IDs and commit once at the end of the transaction to guarantee all-or-nothing database integrity. - Async route testing requires
httpx.AsyncClient: Standard test clients block coroutines; wrapping the FastAPI app inAsyncClient(app=app)delivers true asynchronous end-to-end testing. - CI pipelines must use real service containers: Running unit and integration tests against real PostgreSQL and Redis containers in GitHub Actions eliminates "works on my machine" deployment bugs.
In the final chapter, Part 6, we cover production operations: Single VPS Production Deployment, Multi-Container Docker Compose, Nginx SSL Reverse Proxy, and Automated Health Monitoring & Auto-Recovery Scripts.
