diff --git a/app/api/v1/jobs.py b/app/api/v1/jobs.py index a442c1f..2f452b7 100644 --- a/app/api/v1/jobs.py +++ b/app/api/v1/jobs.py @@ -6,3 +6,63 @@ from app.schemas.job import JobResponse, JobStatusResponse, JobResultResponse from app.services.job_service import JobService router = APIRouter(prefix="/jobs", tags=["jobs"]) + + +@router.post("/upload", response_model=JobResponse, status_code=status.HTTP_201_CREATED) +async def upload_csv( + file: UploadFile = File(...), + service: JobService = Depends(get_job_service), +): + if not file.filename.endswith(".csv"): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Only CSV files are supported.", + ) + + content = await file.read() + try: + job = await service.upload_csv(filename=file.filename, content=content) + return job + except InvalidCSVException as e: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e), + ) + + +@router.get("/{job_id}/status", response_model=JobStatusResponse) +async def get_job_status( + job_id: str, + service: JobService = Depends(get_job_service), +): + try: + job = await service.get_job_status(job_id) + return job + except JobNotFoundException as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) + + +@router.get("/{job_id}/results", response_model=JobResultResponse) +async def get_job_result( + job_id: str, + service: JobService = Depends(get_job_service), +): + try: + job = await service.get_job_results(job_id) + return job + except JobNotFoundException as e: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=str(e), + ) + + +@router.get("", response_model=List[JobResponse]) +async def get_all_jobs( + status: Optional[str] = Query(None, description="Filter jobs by status"), + service: JobService = Depends(get_job_service), +): + return await service.list_jobs(status=status) diff --git a/app/core/config.py b/app/core/config.py index 96e9e8b..2770d75 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -1,4 +1,3 @@ -from typing import Any from pydantic import Field, computed_field from pydantic_settings import BaseSettings, SettingsConfigDict diff --git a/app/repositories/job_repository.py b/app/repositories/job_repository.py index 64baf9f..e40cb83 100644 --- a/app/repositories/job_repository.py +++ b/app/repositories/job_repository.py @@ -1,4 +1,4 @@ -from typing import List, Optional, Any, Dict +from typing import List, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models.job import Job diff --git a/app/schemas/job.py b/app/schemas/job.py index 7558e8d..8677d84 100644 --- a/app/schemas/job.py +++ b/app/schemas/job.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Optional from pydantic import BaseModel, ConfigDict diff --git a/tests/conftest.py b/tests/conftest.py index a46deaf..93e10d1 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,7 +10,9 @@ from app.db.base import Base from app.main import app # Use a test database suffix or a separate database for testing -TEST_DATABASE_URL = settings.DATABASE_URL.replace(settings.POSTGRES_DB, f"test_{settings.POSTGRES_DB}") +TEST_DATABASE_URL = settings.DATABASE_URL.replace( + settings.POSTGRES_DB, f"test_{settings.POSTGRES_DB}" +) # Create async engine for test db test_engine = create_async_engine( @@ -33,7 +35,6 @@ _db_offline = False def initialize_test_db(): """Create test tables and clean them up after all tests finish.""" global _db_offline - import asyncio try: loop = asyncio.get_event_loop_policy().get_event_loop() except RuntimeError: @@ -85,16 +86,17 @@ async def db_session() -> AsyncGenerator[AsyncSession, None]: @pytest.fixture async def client(db_session: AsyncSession) -> AsyncGenerator[AsyncClient, None]: """Provide an HTTPX AsyncClient for FastAPI endpoint testing with db overrides.""" + # Override get_db dependency to use the test session async def override_get_db() -> AsyncGenerator[AsyncSession, None]: yield db_session app.dependency_overrides[get_db] = override_get_db - + async with AsyncClient( transport=ASGITransport(app=app), base_url="http://test", ) as client: yield client - + app.dependency_overrides.clear() diff --git a/tests/repositories/test_job_repository.py b/tests/repositories/test_job_repository.py index b370a07..db2fff2 100644 --- a/tests/repositories/test_job_repository.py +++ b/tests/repositories/test_job_repository.py @@ -1,7 +1,6 @@ import pytest from sqlalchemy.ext.asyncio import AsyncSession from app.repositories.job_repository import JobRepository -from app.db.models.job import Job # Set up asyncio marker pytestmark = pytest.mark.asyncio @@ -9,13 +8,13 @@ pytestmark = pytest.mark.asyncio async def test_create_and_get_job(db_session: AsyncSession): repo = JobRepository(db_session) - + # Create job job = await repo.create(filename="test_transactions.csv") assert job.id is not None assert job.status == "pending" assert job.filename == "test_transactions.csv" - + # Retrieve job fetched_job = await repo.get_by_id(job.id) assert fetched_job is not None @@ -26,13 +25,15 @@ async def test_create_and_get_job(db_session: AsyncSession): async def test_update_job(db_session: AsyncSession): repo = JobRepository(db_session) job = await repo.create(filename="update_test.csv") - + # Update status and count - updated = await repo.update(job, status="completed", row_count=10, summary={"total": 100}) + updated = await repo.update( + job, status="completed", row_count=10, summary={"total": 100} + ) assert updated.status == "completed" assert updated.row_count == 10 assert updated.summary == {"total": 100} - + # Retrieve to verify persistence fetched = await repo.get_by_id(job.id) assert fetched.status == "completed" @@ -41,16 +42,16 @@ async def test_update_job(db_session: AsyncSession): async def test_list_jobs(db_session: AsyncSession): repo = JobRepository(db_session) - + # Clear existing if any await repo.create(filename="list_1.csv") await repo.create(filename="list_2.csv") - + jobs = await repo.list_all() assert len(jobs) >= 2 - + pending_jobs = await repo.list_all(status="pending") assert len(pending_jobs) >= 2 - + completed_jobs = await repo.list_all(status="completed") assert len(completed_jobs) == 1 # From the update test above