Coverage for backend / app / data_tables / routers.py: 90%
41 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-17 21:34 +0000
1"""Module for generating CRUD routers for the JAM data tables"""
3import base64
5from fastapi import Depends, status, HTTPException, Response
6from sqlalchemy.orm import Session
8from app import models, database
9from app.core import oauth2
10from app.data_tables import schemas
11from app.geolocation.geolocation import geocode_location
12from app.routers.utility import generate_data_table_crud_router
14# ---------------------------------------------------- SIMPLE TABLES ---------------------------------------------------
16# Keyword router
17keyword_router = generate_data_table_crud_router(
18 table_model=models.Keyword,
19 create_schema=schemas.KeywordCreate,
20 update_schema=schemas.KeywordUpdate,
21 out_schema=schemas.KeywordOut,
22 endpoint="keywords",
23 not_found_msg="Keyword not found",
24)
26# Aggregator router
27aggregator_router = generate_data_table_crud_router(
28 table_model=models.Aggregator,
29 create_schema=schemas.AggregatorCreate,
30 update_schema=schemas.AggregatorUpdate,
31 out_schema=schemas.AggregatorOut,
32 endpoint="aggregators",
33 not_found_msg="Aggregator not found",
34)
36# Company router
37company_router = generate_data_table_crud_router(
38 table_model=models.Company,
39 create_schema=schemas.CompanyCreate,
40 update_schema=schemas.CompanyUpdate,
41 out_schema=schemas.CompanyOut,
42 endpoint="companies",
43 not_found_msg="Company not found",
44)
47# Location router
48def transform_location(location_data: dict, db: Session, entry_data: dict | None = None) -> dict:
49 """Geolocate the location data before creating/updating the record.
50 :param location_data: The location data dictionary.
51 :param db: The database session.
52 :param entry_data: optional original data of the entry
53 :return: The transformed location data dictionary with geolocation_id set."""
55 if entry_data:
56 location_data = location_data.copy()
57 location_data.update(entry_data)
58 params = {
59 "postcode": location_data.get("postcode"),
60 "city": location_data.get("city"),
61 "country": location_data.get("country"),
62 }
63 geolocation = geocode_location(params, db) if params else None
64 return {"geolocation_id": geolocation.id if geolocation else None}
67location_router = generate_data_table_crud_router(
68 table_model=models.Location,
69 create_schema=schemas.LocationCreate,
70 update_schema=schemas.LocationUpdate,
71 out_schema=schemas.LocationOut,
72 endpoint="locations",
73 not_found_msg="Location not found",
74 transform=transform_location,
75)
78# File router
79file_router = generate_data_table_crud_router(
80 table_model=models.File,
81 create_schema=schemas.FileCreate,
82 update_schema=schemas.FileUpdate,
83 out_schema=schemas.FileOut,
84 endpoint="files",
85 not_found_msg="File not found",
86)
89@file_router.get("/{file_id}/download")
90def download_file(
91 file_id: int,
92 db: Session = Depends(database.get_db),
93 current_user: models.User = Depends(oauth2.get_current_user),
94):
95 """Download a file by ID.
96 :param file_id: The file ID.
97 :param db: The database session.
98 :param current_user: The current user."""
100 # Get file record from the database
101 file_record = (
102 db.query(models.File).filter(models.File.id == file_id, models.File.owner_id == current_user.id).first()
103 )
105 if not file_record:
106 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File not found")
108 if not file_record.content:
109 raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="File content not found")
111 try:
112 # Content is already a base64 string - just decode it
113 if file_record.content.startswith("data:"):
114 encoded_data = file_record.content.split(",", 1)[1]
115 file_content = base64.b64decode(encoded_data)
116 else:
117 # Pure base64 string
118 file_content = base64.b64decode(file_record.content)
120 except Exception as e:
121 raise HTTPException(
122 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=f"Error decoding file content: {str(e)}"
123 )
125 content_type = file_record.type if file_record.type else "application/octet-stream"
127 return Response(
128 content=file_content,
129 media_type=content_type,
130 headers={
131 "Content-Disposition": f'attachment; filename="{file_record.filename}"',
132 "Content-Length": str(len(file_content)),
133 },
134 )
137# --------------------------------------------------- COMPLEX TABLES ---------------------------------------------------
140# Person router
141person_router = generate_data_table_crud_router(
142 table_model=models.Person,
143 create_schema=schemas.PersonCreate,
144 update_schema=schemas.PersonUpdate,
145 out_schema=schemas.PersonOut,
146 endpoint="persons",
147 not_found_msg="Person not found",
148)
150# Job router
151job_router = generate_data_table_crud_router(
152 table_model=models.Job,
153 create_schema=schemas.JobCreate,
154 update_schema=schemas.JobUpdate,
155 out_schema=schemas.JobOut,
156 endpoint="jobs",
157 not_found_msg="Job not found",
158 many_to_many_fields={
159 "keywords": {
160 "table": models.job_keyword_mapping,
161 "local_key": "job_id",
162 "remote_key": "keyword_id",
163 "related_model": models.Keyword,
164 },
165 "contacts": {
166 "table": models.job_contact_mapping,
167 "local_key": "job_id",
168 "remote_key": "person_id",
169 "related_model": models.Person,
170 },
171 },
172)
174# Interview router
175interview_router = generate_data_table_crud_router(
176 table_model=models.Interview,
177 create_schema=schemas.InterviewCreate,
178 update_schema=schemas.InterviewUpdate,
179 out_schema=schemas.InterviewOut,
180 endpoint="interviews",
181 not_found_msg="Interview not found",
182 many_to_many_fields={
183 "interviewers": {
184 "table": models.interview_interviewer_mapping,
185 "local_key": "interview_id",
186 "remote_key": "person_id",
187 "related_model": models.Person,
188 },
189 },
190)
192# Job Application Update router
193job_application_update_router = generate_data_table_crud_router(
194 table_model=models.JobApplicationUpdate,
195 create_schema=schemas.JobApplicationUpdateCreate,
196 update_schema=schemas.JobApplicationUpdateUpdate,
197 out_schema=schemas.JobApplicationUpdateOut,
198 endpoint="job-application-updates",
199 not_found_msg="Job Application Update not found",
200)
202# Speculative Application router
203speculative_application_update_router = generate_data_table_crud_router(
204 table_model=models.SpeculativeApplication,
205 create_schema=schemas.SpeculativeApplicationCreate,
206 update_schema=schemas.SpeculativeApplicationUpdate,
207 out_schema=schemas.SpeculativeApplicationOut,
208 endpoint="speculative-applications",
209 not_found_msg="Speculative Application not found",
210 many_to_many_fields={
211 "contacts": {
212 "table": models.speculative_application_contact_mapping,
213 "local_key": "speculative_application_id",
214 "remote_key": "person_id",
215 "related_model": models.Person,
216 },
217 },
218)