feat: Setup pagination dependency injections

Author: existentialcoder

apps/backend/src/api/deps/pagination.py

@@ -1,6 +1,8 @@
 from fastapi import Query, HTTPException
 from math import ceil
-from typing import Optional
+from typing import List, Optional, Type, Dict
+
+from pydantic import BaseModel
 from ...core.constants import Constants
 
 
@@ -24,8 +26,8 @@ def paginate_query(query, pagination: dict):
     page = pagination.get('page', 1)
     per_page = pagination.get('per_page', Constants.DEFAULT_PAGE_SIZE)
 
-    if per_page > Constants.MAX_PER_PAGE:
-        raise HTTPException(status_code=400, detail=f'per_page cannot exceed {Constants.MAX_PER_PAGE}')
+    if per_page > Constants.MAX_PAGE_SIZE:
+        raise HTTPException(status_code=400, detail=f'per_page cannot exceed {Constants.MAX_PAGE_SIZE}')
 
     offset = (page - 1) * per_page
     return query.limit(per_page).offset(offset)
@@ -48,3 +50,10 @@ def build_paginated_response(items: list, total: int, page: int, per_page: int):
         },
         'results': items,
     }
+
+def get_paginated_response_model(item_model: Type[BaseModel]) -> Type[BaseModel]:
+    class PaginatedResponse(BaseModel):
+        meta: Dict[str, int]
+        results: List[item_model]
+
+    return PaginatedResponse

apps/backend/src/api/v1/routes/jobs.py

@@ -20,7 +20,8 @@
 #         raise HTTPException(status_code=404, detail='Job not found')
 #     return db_job
 
-@router.get('/', response_model=list[schemas.JobBase])
+
+@router.get('/', response_model=job_service.PaginatedJobs)
 def list_jobs(query: str | None = None, pagination: dict = Depends(pagination_params), db: Session = Depends(get_db)):
     return job_service.get_jobs(db, pagination, query)
 

apps/backend/src/models/job.py

@@ -26,13 +26,12 @@ class Job(Base):
 
     years_of_experience: Mapped[dict] = mapped_column(JSON, nullable=True)
 
-    company: Mapped['Company'] = relationship('Company', back_populates='jobs')
+    company: Mapped['Company'] = relationship('Company')
     company_id: Mapped[int] = mapped_column(ForeignKey('companies.id'), nullable=True)
 
     required_skills: Mapped[list['Skill']] = relationship(
         'Skill',
-        secondary=job_skill_table,
-        back_populates='jobs',
+        secondary=job_skill_table
     )
 
     def __repr__(self):

apps/backend/src/services/job.py

@@ -1,18 +1,32 @@
-import sqlalchemy
+from typing import Type
 from sqlalchemy.orm import Session
 from ..models.job import Job
-from ..schemas.job import JobBase
-from ..api.deps.pagination import paginate_query
+from ..schemas.job import JobBase, JobRead
+from ..api.deps.pagination import build_paginated_response, get_paginated_response_model, paginate_query
 
 # allowed_job_query_fields = [JobBase.title, JobBase.description, JobBase.company]
 
-def get_jobs(db: Session, pagination: dict, query: str | None = None) -> list[JobBase]:
+PaginatedJobs = get_paginated_response_model(JobBase)
+
+def get_jobs(db: Session, pagination: dict, query: str | None = None) -> PaginatedJobs:
     # query maybe like this query="company: Acme Inc,location: Remote"
     q = db.query(Job)
 
+    # Optional search filter
     # if query:
-    #     query_filters = []
-    #     for field in allowed_job_query_fields:
-    #         query_filters.append(getattr(JobBase, field).ilike(f'%{query}%'))
-    #     q = q.filter(sqlalchemy.or_(*query_filters))
-    return paginate_query(q, pagination)
+    #     filters = [getattr(Job, field).ilike(f"%{query}%") for field in allowed_job_query_fields]
+    #     q = q.filter(or_(*filters))
+
+    total = q.count()  # total before pagination
+    q = paginate_query(q, pagination)
+
+    results = q.all()
+
+    # convert SQLAlchemy models to Pydantic
+    jobs = [JobRead.from_orm(job) for job in results]
+
+    return build_paginated_response(
+        items=jobs,
+        total=total,
+        **pagination
+    )