Coverage for backend / app / job_email_scraping / routers / service_log.py: 100%

1"""FastAPI routers for the job email scraping service endpoints. 

2 

3Provides REST API endpoints for managing job alert emails, scraped job postings, 

4and service execution logs with CRUD operations and admin access controls.""" 

5 

6import datetime as dt 

7 

8from fastapi import APIRouter, Depends, Query 

9from sqlalchemy.orm import Session 

10 

11from app import models 

12from app.core.oauth2 import get_current_user 

13from app.database import get_db 

14from app.job_email_scraping import schemas 

15from app.service_runner import routers 

16 

17 

18job_scraping_service_log_router = APIRouter(prefix="/job-scraping-service-logs", tags=["job-scraping-service-logs"]) 

19 

20 

21# GET endpoint for admins to get the service logs 

22@job_scraping_service_log_router.get("/", response_model=list[schemas.JobEmailScrapingServiceLogOut]) 

23def get_service_logs_by_date_range( 

24 start_date: dt.datetime | None = Query(None, description="Start date for filtering (ISO format)"), 

25 end_date: dt.datetime | None = Query(None, description="End date for filtering (ISO format)"), 

26 delta_days: int | None = Query(None, description="Number of days to go back in time"), 

27 limit: int | None = Query(None, description="Maximum number of logs to return"), 

28 current_user: models.User = Depends(get_current_user), 

29 db: Session = Depends(get_db), 

30): 

31 """Get service logs within a specified date range. Admin access required. 

32 :param start_date: Optional start date filter (inclusive) 

33 :param end_date: Optional end date filter (inclusive) 

34 :param limit: Optional limit for number of logs to return 

35 :param delta_days: Optional number of days to go back in time 

36 :param current_user: Current authenticated admin user 

37 :param db: Database session 

38 :return: list of service logs within the date range ordered by run_datetime descending""" 

39 

40 return routers.get_service_logs_by_date_range( 

41 start_date, 

42 end_date, 

43 delta_days, 

44 limit, 

45 current_user, 

46 db, 

47 models.JobEmailScrapingServiceLog, 

48 ) 

49 

50 

51# GET endpoint for admin user to get the latest service log 

52@job_scraping_service_log_router.get("/latest", response_model=schemas.JobEmailScrapingServiceLogOut) 

53def get_latest( 

54 current_user: models.User = Depends(get_current_user), 

55 db: Session = Depends(get_db), 

56): 

57 """Get the latest service log entry. Admin access required. 

58 :param current_user: Current authenticated admin user 

59 :param db: Database session 

60 :return: Latest service log entry""" 

61 

62 return routers.get_latest(current_user, db, models.JobEmailScrapingServiceLog)