Coverage for dormatory/api/routes/links.py: 94%
130 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-08-04 00:22 +0000
1"""
2Links API routes for DORMATORY.
4This module provides RESTful API endpoints for managing link entities.
5"""
7from typing import List, Optional
9from fastapi import APIRouter, HTTPException, Depends
10from pydantic import BaseModel
11from sqlalchemy.orm import Session
13from dormatory.api.dependencies import get_db
14from dormatory.models.dormatory_model import Link, Object
16router = APIRouter(tags=["links"])
19class LinkCreate(BaseModel):
20 parent_id: int
21 parent_type: str
22 child_type: str
23 r_name: str
24 child_id: int
27class LinkUpdate(BaseModel):
28 parent_type: Optional[str] = None
29 child_type: Optional[str] = None
30 r_name: Optional[str] = None
33class LinkResponse(BaseModel):
34 id: int
35 parent_id: int
36 parent_type: str
37 child_type: str
38 r_name: str
39 child_id: int
41 class Config:
42 from_attributes = True
45@router.post("/", response_model=LinkResponse)
46async def create_link(link_data: LinkCreate, db: Session = Depends(get_db)):
47 """
48 Create a new link.
50 Args:
51 link_data: Link creation data
52 db: Database session
54 Returns:
55 Created link data
56 """
57 # Verify that the parent object exists
58 parent_obj = db.query(Object).filter(Object.id == link_data.parent_id).first()
59 if not parent_obj:
60 raise HTTPException(status_code=404, detail="Parent object not found")
62 # Verify that the child object exists
63 child_obj = db.query(Object).filter(Object.id == link_data.child_id).first()
64 if not child_obj:
65 raise HTTPException(status_code=404, detail="Child object not found")
67 # Check for circular reference (parent cannot be child of its own child)
68 if link_data.parent_id == link_data.child_id:
69 raise HTTPException(status_code=422, detail="Cannot create self-referencing link")
71 # Check if link already exists
72 existing_link = db.query(Link).filter(
73 Link.parent_id == link_data.parent_id,
74 Link.child_id == link_data.child_id,
75 Link.r_name == link_data.r_name
76 ).first()
78 if existing_link:
79 raise HTTPException(status_code=409, detail="Link already exists")
81 # Create the link
82 db_link = Link(
83 parent_id=link_data.parent_id,
84 parent_type=link_data.parent_type,
85 child_type=link_data.child_type,
86 r_name=link_data.r_name,
87 child_id=link_data.child_id
88 )
90 db.add(db_link)
91 db.commit()
92 db.refresh(db_link)
94 return LinkResponse.from_orm(db_link)
97@router.get("/{link_id}", response_model=LinkResponse)
98async def get_link_by_id(link_id: int, db: Session = Depends(get_db)):
99 """
100 Get a link by its ID.
102 Args:
103 link_id: Link ID
104 db: Database session
106 Returns:
107 Link data
108 """
109 db_link = db.query(Link).filter(Link.id == link_id).first()
110 if not db_link:
111 raise HTTPException(status_code=404, detail="Link not found")
113 return LinkResponse.from_orm(db_link)
116@router.get("/", response_model=List[LinkResponse])
117async def get_all_links(
118 skip: int = 0,
119 limit: int = 100,
120 parent_id: Optional[int] = None,
121 child_id: Optional[int] = None,
122 r_name: Optional[str] = None,
123 db: Session = Depends(get_db)
124):
125 """
126 Get all links with optional filtering.
128 Args:
129 skip: Number of records to skip
130 limit: Maximum number of records to return
131 parent_id: Filter by parent ID
132 child_id: Filter by child ID
133 r_name: Filter by relationship name
134 db: Database session
136 Returns:
137 List of links
138 """
139 query = db.query(Link)
141 # Apply filters
142 if parent_id:
143 query = query.filter(Link.parent_id == parent_id)
144 if child_id:
145 query = query.filter(Link.child_id == child_id)
146 if r_name:
147 query = query.filter(Link.r_name == r_name)
149 # Apply pagination
150 links = query.offset(skip).limit(limit).all()
152 return [LinkResponse.from_orm(link) for link in links]
155@router.put("/{link_id}", response_model=LinkResponse)
156async def update_link(link_id: int, link_data: LinkUpdate, db: Session = Depends(get_db)):
157 """
158 Update an existing link.
160 Args:
161 link_id: Link ID to update
162 link_data: Updated link data
163 db: Database session
165 Returns:
166 Updated link data
167 """
168 db_link = db.query(Link).filter(Link.id == link_id).first()
169 if not db_link:
170 raise HTTPException(status_code=404, detail="Link not found")
172 # Update fields if provided
173 if link_data.parent_type is not None:
174 db_link.parent_type = link_data.parent_type
175 if link_data.child_type is not None:
176 db_link.child_type = link_data.child_type
177 if link_data.r_name is not None:
178 db_link.r_name = link_data.r_name
180 db.commit()
181 db.refresh(db_link)
183 return LinkResponse.from_orm(db_link)
186@router.delete("/{link_id}")
187async def delete_link(link_id: int, db: Session = Depends(get_db)):
188 """
189 Delete a link.
191 Args:
192 link_id: Link ID to delete
193 db: Database session
195 Returns:
196 Success message
197 """
198 db_link = db.query(Link).filter(Link.id == link_id).first()
199 if not db_link:
200 raise HTTPException(status_code=404, detail="Link not found")
202 db.delete(db_link)
203 db.commit()
205 return {"message": "Link deleted successfully"}
208@router.post("/bulk", response_model=List[LinkResponse])
209async def create_links_bulk(link_data: List[LinkCreate], db: Session = Depends(get_db)):
210 """
211 Create multiple links in a single operation.
213 Args:
214 link_data: List of link creation data
215 db: Database session
217 Returns:
218 List of created links
219 """
220 created_links = []
222 for item in link_data:
223 # Verify that the parent object exists
224 parent_obj = db.query(Object).filter(Object.id == item.parent_id).first()
225 if not parent_obj:
226 raise HTTPException(status_code=404, detail=f"Parent object {item.parent_id} not found")
228 # Verify that the child object exists
229 child_obj = db.query(Object).filter(Object.id == item.child_id).first()
230 if not child_obj:
231 raise HTTPException(status_code=404, detail=f"Child object {item.child_id} not found")
233 # Check for circular reference
234 if item.parent_id == item.child_id:
235 raise HTTPException(status_code=422, detail="Cannot create self-referencing link")
237 # Check if link already exists
238 existing_link = db.query(Link).filter(
239 Link.parent_id == item.parent_id,
240 Link.child_id == item.child_id,
241 Link.r_name == item.r_name
242 ).first()
244 if existing_link:
245 raise HTTPException(status_code=409, detail=f"Link already exists between parent {item.parent_id} and child {item.child_id}")
247 # Create the link
248 db_link = Link(
249 parent_id=item.parent_id,
250 parent_type=item.parent_type,
251 child_type=item.child_type,
252 r_name=item.r_name,
253 child_id=item.child_id
254 )
256 db.add(db_link)
257 created_links.append(db_link)
259 db.commit()
261 # Refresh all links to get their IDs
262 for link in created_links:
263 db.refresh(link)
265 return [LinkResponse.from_orm(link) for link in created_links]
268@router.get("/parent/{parent_id}/children")
269async def get_children_by_parent(parent_id: int, db: Session = Depends(get_db)):
270 """
271 Get all children of a parent object.
273 Args:
274 parent_id: Parent object ID
275 db: Database session
277 Returns:
278 List of child objects
279 """
280 # Verify that the parent object exists
281 parent_obj = db.query(Object).filter(Object.id == parent_id).first()
282 if not parent_obj:
283 raise HTTPException(status_code=404, detail="Parent object not found")
285 # Get all links where this object is the parent
286 links = db.query(Link).filter(Link.parent_id == parent_id).all()
288 # Get the child objects
289 child_ids = [link.child_id for link in links]
290 children = db.query(Object).filter(Object.id.in_(child_ids)).all()
292 return [
293 {
294 "id": child.id,
295 "name": child.name,
296 "type_id": str(child.type_id),
297 "relationship": next(link.r_name for link in links if link.child_id == child.id)
298 }
299 for child in children
300 ]
303@router.get("/child/{child_id}/parents")
304async def get_parents_by_child(child_id: int, db: Session = Depends(get_db)):
305 """
306 Get all parents of a child object.
308 Args:
309 child_id: Child object ID
310 db: Database session
312 Returns:
313 List of parent objects
314 """
315 # Verify that the child object exists
316 child_obj = db.query(Object).filter(Object.id == child_id).first()
317 if not child_obj:
318 raise HTTPException(status_code=404, detail="Child object not found")
320 # Get all links where this object is the child
321 links = db.query(Link).filter(Link.child_id == child_id).all()
323 # Get the parent objects
324 parent_ids = [link.parent_id for link in links]
325 parents = db.query(Object).filter(Object.id.in_(parent_ids)).all()
327 return [
328 {
329 "id": parent.id,
330 "name": parent.name,
331 "type_id": str(parent.type_id),
332 "relationship": next(link.r_name for link in links if link.parent_id == parent.id)
333 }
334 for parent in parents
335 ]
338@router.get("/relationship/{r_name}")
339async def get_links_by_relationship(r_name: str, db: Session = Depends(get_db)):
340 """
341 Get all links with a specific relationship name.
343 Args:
344 r_name: Relationship name
345 db: Database session
347 Returns:
348 List of links with this relationship
349 """
350 links = db.query(Link).filter(Link.r_name == r_name).all()
352 return [LinkResponse.from_orm(link) for link in links]
355@router.post("/hierarchy")
356async def create_hierarchy(hierarchy_data: dict, db: Session = Depends(get_db)):
357 """
358 Create a complete hierarchy structure.
360 Args:
361 hierarchy_data: Hierarchy creation data
362 db: Database session
364 Returns:
365 Created hierarchy structure
366 """
367 # TODO: Implement hierarchy creation
368 raise HTTPException(status_code=500, detail="Not implemented")