Merge pull request #1161 from Benji918/feat/filtered_job_search

feat: implemeted filtered job search endpoint

Author: joboy-dev

api/v1/routes/jobs.py

@@ -4,7 +4,7 @@
 from api.v1.schemas.jobs import PostJobSchema, AddJobSchema, JobCreateResponseSchema, UpdateJobSchema
 from fastapi.exceptions import HTTPException
 from fastapi.encoders import jsonable_encoder
-from typing import Annotated
+from typing import Annotated, Optional
 from fastapi import APIRouter, HTTPException, Depends, status, Query
 
 from api.v1.services.user import user_service
@@ -60,9 +60,38 @@ async def add_jobs(
         data=jsonable_encoder(JobCreateResponseSchema.model_validate(new_job))
     )
 
+@jobs.get("/filter", response_model=success_response)
+async def filter(
+        title: Optional[str] = None,
+        location: Optional[str] = None,
+        job_type: Optional[str] = None,
+        db: Session = Depends(get_db)
+):
+    """
+        Retrieve job details by specified search parameters salary range, location and job_type.
+        This endpoint to handle job filtering based on user preferences. This endpoint will allow users to filter
+        job listings by parameters such as salary range, location, and job type to find positions that match
+        their specific needs
+
+        Parameters:
+        - title: str (optional)
+            The job title
+        - location: str (optional)
+            The job location
+        - job_type: str (optional)
+            The type of job
+        - db: The database session
+        """
+    jobs = job_service.fetch_by_filters(db, title, location, job_type)
+
+    return success_response(
+        status_code=status.HTTP_200_OK,
+        data=jsonable_encoder(jobs),
+        message= f"Successfully retrieved {len(jobs)} jobs"
+    )
 
 @jobs.get("/{job_id}", response_model=success_response)
-async def get_job(
+async def retrieveJob(
     job_id: str,
     db: Session = Depends(get_db)
 ):
@@ -75,15 +104,14 @@ async def get_job(
         The ID of the job to retrieve.
     - db: The database session
     """
-    job = job_service.fetch(db, job_id)
+    job = job_service.retrieve(db, job_id)
 
     return success_response(
         message="Retrieved Job successfully",
-        status_code=200,
+        status_code=status.HTTP_200_OK,
         data=jsonable_encoder(job)
     )
 
-
 @jobs.get("")
 async def fetch_all_jobs(
     db: Session = Depends(get_db),
@@ -146,6 +174,7 @@ async def update_job(
     )
 
 
+
 # -------------------- JOB APPLICATION ROUTES ------------------------
 # --------------------------------------------------------------------
 
@@ -211,7 +240,7 @@ async def fetch_all_job_applications(
     Args:
         - job_id (str): The Job ID
         - db (Annotated[Session, Depends): the database session
-        - current_user: The current authenticated super admin 
+        - current_user: The current authenticated super admin
         - per_page: Number of customers per page (default: 10, minimum: 1)
         - page: Page number (starts from 1)
 
@@ -242,6 +271,7 @@ async def delete_application(job_id: str,
     job_application_service.delete(job_id, application_id, db)
 
 
+
 # -------------------- JOB BOOKMARK ROUTES ------------------------
 # --------------------------------------------------------------------
 
@@ -277,7 +307,7 @@ async def create_bookmark(
             "status": "failure",
             "message": "Job not listed",
             "status_code": 400,
-            "data": {} 
+            "data": {}
         }
     except HTTPException as e:
         return {

api/v1/services/jobs.py

@@ -4,9 +4,10 @@
 from api.core.base.services import Service
 from api.v1.models.job import Job
 from fastapi import HTTPException
+from api.utils.db_validators import check_model_existence
 
 
-class JobService(Service):
+class JobService():
     """Job service functionality"""
 
     def create(self, db: Session, schema) -> Job:
@@ -32,20 +33,35 @@ def fetch_all(self, db: Session, **query_params: Optional[Any]):
 
         return query.all()
 
-
-    @staticmethod
-    def fetch(db: Session, id: str) -> Optional[Job]:
+    def retrieve(self, db: Session, job_id):
         """Fetches a job by ID"""
-        return db.query(Job).filter(Job.id == id).first()
-
-    def fetch(self, db: Session, id: str):
-        """Fetches a job by id"""
-        job = db.query(Job).filter(Job.id == id).first()
+        job = db.query(Job).filter(Job.id == job_id).first()
         if not job:
             raise HTTPException(status_code=404, detail="Job not found")
         return job
 
 
+    def fetch_by_filters(self, db: Session, title: str = None, location: str = None, job_type: str = None):
+        """Fetch jobs by the specified filters"""
+        query = db.query(Job)
+
+        if title:
+            query = query.filter(Job.title.ilike(f"%{title}%"))
+        if location:
+            query = query.filter(Job.location.ilike(f"%{location}%"))
+        if job_type:
+            query = query.filter(Job.job_type == job_type)
+
+        jobs = query.all()
+
+        if not jobs:
+            raise HTTPException(
+                status_code=404,
+                detail="No jobs found matching the search parameters."
+            )
+        return jobs
+
+
     def update(self, db: Session, id: str, schema):
         """Updates a job"""
 

tests/v1/job/test_filter_jobs.py

@@ -0,0 +1,116 @@
+import pytest
+from fastapi.testclient import TestClient
+from main import app
+from api.db.database import get_db
+from api.v1.models.job import Job
+from faker import Faker
+
+fake = Faker()
+client = TestClient(app)
+
+class FakeJob:
+    def __init__(self, title, location, job_type):
+        self.title = title
+        self.location = location
+        self.job_type = job_type
+    def dict(self):
+        return {"title": self.title, "location": self.location, "job_type": self.job_type}
+
+class FakeQuery:
+    def __init__(self, jobs):
+        self.jobs = jobs
+    def filter(self, predicate):
+        filtered = list(filter(predicate, self.jobs))
+        return FakeQuery(filtered)
+    def all(self):
+        return self.jobs
+
+class FakeSession:
+    def __init__(self, jobs):
+        self.jobs = jobs
+    def query(self, model):
+        return FakeQuery(self.jobs)
+
+class FakeColumn:
+    def __init__(self, attr_name):
+        self.attr_name = attr_name
+    def ilike(self, pattern):
+        def predicate(job):
+            value = getattr(job, self.attr_name, "")
+            return pattern.strip("%").lower() in value.lower()
+        return predicate
+    def __eq__(self, other):
+        def predicate(job):
+            return getattr(job, self.attr_name, None) == other
+        return predicate
+
+Job.title = FakeColumn("title")
+Job.location = FakeColumn("location")
+Job.job_type = FakeColumn("job_type")
+
+fake_jobs = [
+    FakeJob("Software Engineer", "New York", "Full Time"),
+    FakeJob("Data Scientist", "San Francisco", "Part Time"),
+    FakeJob("Backend Developer", "New York", "Contract")
+]
+
+@pytest.fixture
+def fake_db():
+    yield FakeSession(fake_jobs)
+
+@pytest.fixture(autouse=True)
+def override_get_db(fake_db):
+    def get_db_override():
+        yield fake_db
+    app.dependency_overrides[get_db] = get_db_override
+    yield
+    app.dependency_overrides = {}
+
+def test_filter_no_parameters():
+    response = client.get("/api/v1/jobs/filter")
+    assert response.status_code == 200
+    json_data = response.json()
+    assert json_data["status_code"] == 200
+    assert len(json_data["data"]) == 3
+    assert json_data["message"] == "Successfully retrieved 3 jobs"
+
+def test_filter_by_title():
+    response = client.get("/api/v1/jobs/filter", params={"title": "Software"})
+    assert response.status_code == 200
+    json_data = response.json()
+    assert len(json_data["data"]) == 1
+    assert json_data["data"][0]["title"] == "Software Engineer"
+    assert json_data["message"] == "Successfully retrieved 1 jobs"
+
+def test_filter_by_location():
+    response = client.get("/api/v1/jobs/filter", params={"location": "New York"})
+    assert response.status_code == 200
+    json_data = response.json()
+    assert len(json_data["data"]) == 2
+    titles = [job["title"] for job in json_data["data"]]
+    assert "Software Engineer" in titles
+    assert "Backend Developer" in titles
+    assert json_data["message"] == "Successfully retrieved 2 jobs"
+
+def test_filter_by_job_type():
+    response = client.get("/api/v1/jobs/filter", params={"job_type": "Part Time"})
+    assert response.status_code == 200
+    json_data = response.json()
+    assert len(json_data["data"]) == 1
+    assert json_data["data"][0]["job_type"] == "Part Time"
+    assert json_data["message"] == "Successfully retrieved 1 jobs"
+
+def test_filter_by_multiple_parameters():
+    response = client.get("/api/v1/jobs/filter", params={"title": "Developer", "location": "New York"})
+    assert response.status_code == 200
+    json_data = response.json()
+    assert len(json_data["data"]) == 1
+    assert json_data["data"][0]["title"] == "Backend Developer"
+    assert json_data["data"][0]["location"] == "New York"
+    assert json_data["message"] == "Successfully retrieved 1 jobs"
+
+def test_filter_no_match():
+    response = client.get("/api/v1/jobs/filter", params={"title": "Manager"})
+    assert response.status_code == 404
+    json_data = response.json()
+    assert json_data["message"] == "No jobs found matching the search parameters."