From eb5e3cdc6c0981d3e02d8f9baf95cc21f3e3f1de Mon Sep 17 00:00:00 2001 From: existentialcoder Date: Fri, 31 Oct 2025 23:15:57 +0100 Subject: [PATCH] feat: Setup pagination dependency injections --- apps/backend/src/api/deps/pagination.py | 15 +++++++++--- apps/backend/src/api/v1/routes/jobs.py | 3 ++- apps/backend/src/models/job.py | 5 ++-- apps/backend/src/services/job.py | 32 ++++++++++++++++++------- 4 files changed, 39 insertions(+), 16 deletions(-) diff --git a/apps/backend/src/api/deps/pagination.py b/apps/backend/src/api/deps/pagination.py index b5b37f5..4d704bf 100644 --- a/apps/backend/src/api/deps/pagination.py +++ b/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 diff --git a/apps/backend/src/api/v1/routes/jobs.py b/apps/backend/src/api/v1/routes/jobs.py index 6f1c002..4bbb4b7 100644 --- a/apps/backend/src/api/v1/routes/jobs.py +++ b/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) diff --git a/apps/backend/src/models/job.py b/apps/backend/src/models/job.py index 9f578c0..e9ca0e2 100644 --- a/apps/backend/src/models/job.py +++ b/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): diff --git a/apps/backend/src/services/job.py b/apps/backend/src/services/job.py index 9e4b1f4..336e2b9 100644 --- a/apps/backend/src/services/job.py +++ b/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 + )