Coverage for src/cstlcore/glossary/router.py: 24%
102 statements
« prev ^ index » next coverage.py v7.9.1, created at 2026-02-19 12:46 +0000
1import uuid
3from fastapi import APIRouter, Depends, HTTPException, Query
4from loguru import logger
5from sqlmodel import Session, select
6from rapidfuzz import fuzz
8from cstlcore.constellations.models import Constellation
9from cstlcore.database.dependencies import get_session
10from cstlcore.glossary.models import (
11 GlossaryTerm,
12 GlossaryTermCreate,
13 GlossaryTermPublic,
14 GlossaryTermUpdate,
15)
16from cstlcore.memberships.dependencies import require_read_access, require_write_access
18router = APIRouter()
21@router.post(
22 "/constellations/{constellation_id}/glossary",
23 response_model=GlossaryTermPublic,
24 status_code=201,
25)
26async def create_glossary_term(
27 term_data: GlossaryTermCreate,
28 constellation: Constellation = Depends(require_write_access),
29 session: Session = Depends(get_session),
30) -> GlossaryTerm:
31 """
32 Create a new glossary term.
33 """
34 logger.info(
35 f"Creating glossary term '{term_data.term}' for constellation {constellation.id}"
36 )
38 # Check that the term does not already exist in this constellation
39 statement = select(GlossaryTerm).where(
40 GlossaryTerm.constellation_id == constellation.id,
41 GlossaryTerm.term == term_data.term,
42 )
43 existing_term = session.exec(statement).first()
44 if existing_term:
45 raise HTTPException(
46 status_code=409,
47 detail=f"Term '{term_data.term}' already exists in this constellation",
48 )
50 # Validate that synonym and related term IDs exist within the constellation
51 all_related_ids = set(term_data.synonym_ids + term_data.related_term_ids)
52 if all_related_ids:
53 existing_ids = set(
54 str(term.id)
55 for term in session.exec(
56 select(GlossaryTerm).where(
57 GlossaryTerm.constellation_id == constellation.id,
58 GlossaryTerm.id.in_([uuid.UUID(id) for id in all_related_ids]),
59 )
60 ).all()
61 )
62 invalid_ids = all_related_ids - existing_ids
63 if invalid_ids:
64 raise HTTPException(
65 status_code=400,
66 detail=f"Invalid term IDs: {', '.join(invalid_ids)}",
67 )
69 # Create the term
70 new_term = GlossaryTerm.model_validate(
71 term_data, update={"constellation_id": constellation.id}
72 )
73 session.add(new_term)
74 session.commit()
75 session.refresh(new_term)
77 logger.info(f"Glossary term {new_term.id} created successfully")
78 return new_term
81@router.get(
82 "/constellations/{constellation_id}/glossary",
83 response_model=list[GlossaryTermPublic],
84)
85async def get_glossary_terms(
86 constellation: Constellation = Depends(require_read_access),
87 session: Session = Depends(get_session),
88) -> list[GlossaryTerm]:
89 """
90 Retrieve all glossary terms for a constellation.
91 """
92 logger.info(f"Fetching glossary terms for constellation {constellation.id}")
94 statement = (
95 select(GlossaryTerm)
96 .where(GlossaryTerm.constellation_id == constellation.id)
97 .order_by(GlossaryTerm.term)
98 )
99 terms = session.exec(statement).all()
101 return terms
104@router.get(
105 "/constellations/{constellation_id}/glossary/search",
106 response_model=list[GlossaryTermPublic],
107)
108async def search_glossary_terms(
109 search_query: str = Query(..., description="The search query"),
110 limit: int = Query(default=100, ge=0, description="The maximum number of items to return"),
111 page: int = Query(default=1, ge=1, description="The page number to return"),
112 constellation: Constellation = Depends(require_read_access),
113 session: Session = Depends(get_session),
114) -> list[GlossaryTerm]:
115 """
116 Search glossary terms within a constellation by term name, term definition, or examples (case-insensitive, partial match/fuzzy).
117 """
118 logger.info(
119 f"Searching glossary terms in constellation {constellation.id} for query '{search_query}'"
120 )
122 # Get all terms from the DB
123 all_terms = session.exec(
124 select(GlossaryTerm).where(
125 GlossaryTerm.constellation_id == constellation.id
126 )
127 ).all()
129 # Then filter search using fuzzy matching (fuzz.partial_ratio >= 80)
130 matched_terms = []
131 for term in all_terms:
132 combined_text = f"{term.term} {term.definition or ''} {' '.join(term.examples or [])}"
133 score = fuzz.partial_ratio(search_query.lower(), combined_text.lower())
134 if score >= 80:
135 matched_terms.append((term, score))
137 # Sort by score descending
138 matched_terms.sort(key=lambda x: x[1], reverse=True)
140 # Paginate
141 start_index = (page - 1) * limit
142 end_index = start_index + limit
143 if start_index >= len(matched_terms):
144 return []
145 paginated_terms = [term for term, score in matched_terms[start_index:end_index]]
147 return paginated_terms
150@router.get(
151 "/constellations/{constellation_id}/glossary/{term_id}",
152 response_model=GlossaryTermPublic,
153)
154async def get_glossary_term(
155 term_id: uuid.UUID,
156 constellation: Constellation = Depends(require_read_access),
157 session: Session = Depends(get_session),
158) -> GlossaryTerm:
159 """
160 Retrieve a specific glossary term.
161 """
162 logger.info(f"Fetching glossary term {term_id}")
164 term = session.exec(
165 select(GlossaryTerm).where(
166 GlossaryTerm.id == term_id,
167 GlossaryTerm.constellation_id == constellation.id,
168 )
169 ).first()
170 if not term:
171 raise HTTPException(status_code=404, detail="Glossary term not found")
173 return term
176@router.patch(
177 "/constellations/{constellation_id}/glossary/{term_id}",
178 response_model=GlossaryTermPublic,
179)
180async def update_glossary_term(
181 term_id: uuid.UUID,
182 term_update: GlossaryTermUpdate,
183 constellation: Constellation = Depends(require_write_access),
184 session: Session = Depends(get_session),
185) -> GlossaryTerm:
186 """
187 Update a glossary term.
188 """
189 logger.info(f"Updating glossary term {term_id}")
191 term = session.exec(
192 select(GlossaryTerm).where(
193 GlossaryTerm.id == term_id,
194 GlossaryTerm.constellation_id == constellation.id,
195 )
196 ).first()
197 if not term:
198 raise HTTPException(status_code=404, detail="Glossary term not found")
200 # Validate that synonym and related term IDs exist within the constellation
201 update_data = term_update.model_dump(exclude_unset=True)
202 if "synonym_ids" in update_data or "related_term_ids" in update_data:
203 all_related_ids = set(
204 update_data.get("synonym_ids", []) + update_data.get("related_term_ids", [])
205 )
206 if all_related_ids:
207 existing_ids = set(
208 str(t.id)
209 for t in session.exec(
210 select(GlossaryTerm).where(
211 GlossaryTerm.constellation_id == constellation.id,
212 GlossaryTerm.id.in_([uuid.UUID(id) for id in all_related_ids]),
213 )
214 ).all()
215 )
216 invalid_ids = all_related_ids - existing_ids
217 if invalid_ids:
218 raise HTTPException(
219 status_code=400,
220 detail=f"Invalid term IDs: {', '.join(invalid_ids)}",
221 )
223 # Update the provided fields
224 term.sqlmodel_update(update_data)
226 session.add(term)
227 session.commit()
228 session.refresh(term)
230 logger.info(f"Glossary term {term_id} updated successfully")
231 return term
234@router.delete("/constellations/{constellation_id}/glossary/{term_id}")
235async def delete_glossary_term(
236 term_id: uuid.UUID,
237 constellation: Constellation = Depends(require_write_access),
238 session: Session = Depends(get_session),
239):
240 """
241 Delete a glossary term.
242 """
243 logger.info(f"Deleting glossary term {term_id}")
245 term = session.exec(
246 select(GlossaryTerm).where(
247 GlossaryTerm.id == term_id,
248 GlossaryTerm.constellation_id == constellation.id,
249 )
250 ).first()
251 if not term:
252 raise HTTPException(status_code=404, detail="Glossary term not found")
254 session.delete(term)
255 session.commit()
257 logger.info(f"Glossary term {term_id} deleted successfully")
258 return {"ok": True}
261@router.get(
262 "/constellations/{constellation_id}/glossary/{term_id}/related",
263 response_model=dict[str, list[GlossaryTermPublic]],
264)
265async def get_related_terms(
266 term_id: uuid.UUID,
267 constellation: Constellation = Depends(require_read_access),
268 session: Session = Depends(get_session),
269):
270 """
271 Get the related terms (synonyms and related_terms) of a term.
272 Return a dictionary with keys 'synonyms' and 'related_terms'.
273 """
274 logger.info(f"Fetching related terms for glossary term {term_id}")
276 term = session.exec(
277 select(GlossaryTerm).where(
278 GlossaryTerm.id == term_id,
279 GlossaryTerm.constellation_id == constellation.id,
280 )
281 ).first()
282 if not term:
283 raise HTTPException(status_code=404, detail="Glossary term not found")
285 # Retrieve synonyms
286 synonyms = []
287 if term.synonym_ids:
288 synonyms = list(
289 session.exec(
290 select(GlossaryTerm).where(
291 GlossaryTerm.id.in_([uuid.UUID(id) for id in term.synonym_ids]),
292 GlossaryTerm.constellation_id == constellation.id,
293 )
294 ).all()
295 )
297 # Retrieve related terms
298 related_terms: list[GlossaryTermPublic] = []
299 if term.related_term_ids:
300 related_terms = list(
301 session.exec(
302 select(GlossaryTerm).where(
303 GlossaryTerm.id.in_(
304 [uuid.UUID(id) for id in term.related_term_ids]
305 ),
306 GlossaryTerm.constellation_id == constellation.id,
307 )
308 ).all()
309 )
311 return {"synonyms": synonyms, "related_terms": related_terms}