Coverage for dormatory/api/routes/objects.py: 92%
158 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-08-04 00:22 +0000
1"""
2Objects API routes for DORMATORY.
4This module provides RESTful API endpoints for managing object entities.
5"""
7from typing import List, Optional
8from uuid import UUID, uuid4
10from fastapi import APIRouter, HTTPException, Depends
11from pydantic import BaseModel, Field
12from sqlalchemy.orm import Session
14from dormatory.api.dependencies import get_db
15from dormatory.models.dormatory_model import Object, Type, Link
17router = APIRouter(tags=["objects"])
19class ObjectCreate(BaseModel):
20 name: str = Field(..., min_length=1, description="Object name (cannot be empty)")
21 version: Optional[int] = 1
22 type_id: UUID
23 created_on: str
24 created_by: str
27class ObjectUpdate(BaseModel):
28 name: Optional[str] = None
29 version: Optional[int] = None
30 type_id: Optional[UUID] = None
33class ObjectResponse(BaseModel):
34 id: int
35 name: str
36 version: int
37 type_id: UUID
38 created_on: str
39 created_by: str
41 class Config:
42 from_attributes = True
45@router.post("/", response_model=ObjectResponse)
46async def create_object(object_data: ObjectCreate, db: Session = Depends(get_db)):
47 """
48 Create a new object.
50 Args:
51 object_data: Object creation data
52 db: Database session
54 Returns:
55 Created object data
56 """
57 # Verify that the type exists
58 type_obj = db.query(Type).filter(Type.id == object_data.type_id).first()
59 if not type_obj:
60 raise HTTPException(status_code=404, detail="Type not found")
62 # Create the object
63 db_object = Object(
64 name=object_data.name,
65 version=object_data.version or 1,
66 type_id=object_data.type_id,
67 created_on=object_data.created_on,
68 created_by=object_data.created_by
69 )
71 db.add(db_object)
72 db.commit()
73 db.refresh(db_object)
75 return ObjectResponse.from_orm(db_object)
78@router.get("/{object_id}", response_model=ObjectResponse)
79async def get_object_by_id(object_id: int, db: Session = Depends(get_db)):
80 """
81 Get an object by its ID.
83 Args:
84 object_id: Object ID
85 db: Database session
87 Returns:
88 Object data
89 """
90 db_object = db.query(Object).filter(Object.id == object_id).first()
91 if not db_object:
92 raise HTTPException(status_code=404, detail="Object not found")
94 return ObjectResponse.from_orm(db_object)
97@router.get("/", response_model=List[ObjectResponse])
98async def get_all_objects(
99 skip: int = 0,
100 limit: int = 100,
101 name: Optional[str] = None,
102 type_id: Optional[UUID] = None,
103 db: Session = Depends(get_db)
104):
105 """
106 Get all objects with optional filtering.
108 Args:
109 skip: Number of records to skip
110 limit: Maximum number of records to return
111 name: Filter by object name
112 type_id: Filter by type ID
113 db: Database session
115 Returns:
116 List of objects
117 """
118 query = db.query(Object)
120 # Apply filters
121 if name:
122 query = query.filter(Object.name.contains(name))
123 if type_id:
124 query = query.filter(Object.type_id == type_id)
126 # Apply pagination
127 objects = query.offset(skip).limit(limit).all()
129 return [ObjectResponse.from_orm(obj) for obj in objects]
132@router.put("/{object_id}", response_model=ObjectResponse)
133async def update_object(object_id: int, object_data: ObjectUpdate, db: Session = Depends(get_db)):
134 """
135 Update an existing object.
137 Args:
138 object_id: Object ID to update
139 object_data: Updated object data
140 db: Database session
142 Returns:
143 Updated object data
144 """
145 db_object = db.query(Object).filter(Object.id == object_id).first()
146 if not db_object:
147 raise HTTPException(status_code=404, detail="Object not found")
149 # Update fields if provided
150 if object_data.name is not None:
151 db_object.name = object_data.name
152 if object_data.version is not None:
153 db_object.version = object_data.version
154 if object_data.type_id is not None:
155 # Verify that the new type exists
156 type_obj = db.query(Type).filter(Type.id == object_data.type_id).first()
157 if not type_obj:
158 raise HTTPException(status_code=404, detail="Type not found")
159 db_object.type_id = object_data.type_id
161 db.commit()
162 db.refresh(db_object)
164 return ObjectResponse.from_orm(db_object)
167@router.delete("/{object_id}")
168async def delete_object(object_id: int, db: Session = Depends(get_db)):
169 """
170 Delete an object.
172 Args:
173 object_id: Object ID to delete
174 db: Database session
176 Returns:
177 Success message
178 """
179 db_object = db.query(Object).filter(Object.id == object_id).first()
180 if not db_object:
181 raise HTTPException(status_code=404, detail="Object not found")
183 db.delete(db_object)
184 db.commit()
186 return {"message": "Object deleted successfully"}
189@router.post("/bulk", response_model=List[ObjectResponse])
190async def create_objects_bulk(object_data: List[ObjectCreate], db: Session = Depends(get_db)):
191 """
192 Create multiple objects in a single operation.
194 Args:
195 object_data: List of object creation data
196 db: Database session
198 Returns:
199 List of created objects
200 """
201 created_objects = []
203 for item in object_data:
204 # Verify that the type exists
205 type_obj = db.query(Type).filter(Type.id == item.type_id).first()
206 if not type_obj:
207 raise HTTPException(status_code=404, detail=f"Type {item.type_id} not found")
209 # Create the object
210 db_object = Object(
211 name=item.name,
212 version=item.version or 1,
213 type_id=item.type_id,
214 created_on=item.created_on,
215 created_by=item.created_by
216 )
218 db.add(db_object)
219 created_objects.append(db_object)
221 db.commit()
223 # Refresh all objects to get their IDs
224 for obj in created_objects:
225 db.refresh(obj)
227 return [ObjectResponse.from_orm(obj) for obj in created_objects]
230@router.get("/{object_id}/children")
231async def get_object_children(object_id: int, db: Session = Depends(get_db)):
232 """
233 Get all children of an object.
235 Args:
236 object_id: Object ID
237 db: Database session
239 Returns:
240 List of child objects with relationship information
241 """
242 # Verify that the object exists
243 db_object = db.query(Object).filter(Object.id == object_id).first()
244 if not db_object:
245 raise HTTPException(status_code=404, detail="Object not found")
247 # Get all links where this object is the parent
248 links = db.query(Link).filter(Link.parent_id == object_id).all()
250 # Get the child objects
251 child_ids = [link.child_id for link in links]
252 children = db.query(Object).filter(Object.id.in_(child_ids)).all()
254 return [
255 {
256 "id": child.id,
257 "name": child.name,
258 "version": child.version,
259 "type_id": str(child.type_id),
260 "created_on": child.created_on,
261 "created_by": child.created_by,
262 "relationship": next(link.r_name for link in links if link.child_id == child.id)
263 }
264 for child in children
265 ]
268@router.get("/{object_id}/parents")
269async def get_object_parents(object_id: int, db: Session = Depends(get_db)):
270 """
271 Get all parents of an object.
273 Args:
274 object_id: Object ID
275 db: Database session
277 Returns:
278 List of parent objects with relationship information
279 """
280 # Verify that the object exists
281 db_object = db.query(Object).filter(Object.id == object_id).first()
282 if not db_object:
283 raise HTTPException(status_code=404, detail="Object not found")
285 # Get all links where this object is the child
286 links = db.query(Link).filter(Link.child_id == object_id).all()
288 # Get the parent objects
289 parent_ids = [link.parent_id for link in links]
290 parents = db.query(Object).filter(Object.id.in_(parent_ids)).all()
292 return [
293 {
294 "id": parent.id,
295 "name": parent.name,
296 "version": parent.version,
297 "type_id": str(parent.type_id),
298 "created_on": parent.created_on,
299 "created_by": parent.created_by,
300 "relationship": next(link.r_name for link in links if link.parent_id == parent.id)
301 }
302 for parent in parents
303 ]
306@router.get("/{object_id}/hierarchy")
307async def get_object_hierarchy(object_id: int, db: Session = Depends(get_db)):
308 """
309 Get the complete hierarchy for an object.
311 Args:
312 object_id: Object ID
313 db: Database session
315 Returns:
316 Complete hierarchy tree
317 """
318 # Verify that the object exists
319 db_object = db.query(Object).filter(Object.id == object_id).first()
320 if not db_object:
321 raise HTTPException(status_code=404, detail="Object not found")
323 def build_hierarchy(obj_id: int, visited: set = None) -> dict:
324 """Recursively build the hierarchy tree."""
325 if visited is None:
326 visited = set()
328 if obj_id in visited:
329 return None # Prevent circular references
331 visited.add(obj_id)
333 # Get the object
334 obj = db.query(Object).filter(Object.id == obj_id).first()
335 if not obj:
336 return None
338 # Get children
339 child_links = db.query(Link).filter(Link.parent_id == obj_id).all()
340 children = []
342 for link in child_links:
343 child_hierarchy = build_hierarchy(link.child_id, visited.copy())
344 if child_hierarchy:
345 children.append({
346 "object": child_hierarchy,
347 "relationship": link.r_name
348 })
350 return {
351 "id": obj.id,
352 "name": obj.name,
353 "version": obj.version,
354 "type_id": str(obj.type_id),
355 "created_on": obj.created_on,
356 "created_by": obj.created_by,
357 "children": children
358 }
360 hierarchy = build_hierarchy(object_id)
361 return hierarchy
364@router.get("/{object_id}/hierarchy/{depth}")
365async def get_object_hierarchy_with_depth(object_id: int, depth: int, db: Session = Depends(get_db)):
366 """
367 Get the hierarchy for an object up to a specific depth.
369 Args:
370 object_id: Object ID
371 depth: Maximum depth to retrieve
372 db: Database session
374 Returns:
375 Hierarchy tree up to specified depth
376 """
377 if depth < 0:
378 raise HTTPException(status_code=422, detail="Depth must be non-negative")
380 # Verify that the object exists
381 db_object = db.query(Object).filter(Object.id == object_id).first()
382 if not db_object:
383 raise HTTPException(status_code=404, detail="Object not found")
385 def build_hierarchy_with_depth(obj_id: int, current_depth: int, visited: set = None) -> dict:
386 """Recursively build the hierarchy tree up to specified depth."""
387 if visited is None:
388 visited = set()
390 if obj_id in visited or current_depth > depth:
391 return None # Prevent circular references or exceed depth limit
393 visited.add(obj_id)
395 # Get the object
396 obj = db.query(Object).filter(Object.id == obj_id).first()
397 if not obj:
398 return None
400 # Get children if we haven't reached the depth limit
401 children = []
402 if current_depth < depth:
403 child_links = db.query(Link).filter(Link.parent_id == obj_id).all()
405 for link in child_links:
406 child_hierarchy = build_hierarchy_with_depth(link.child_id, current_depth + 1, visited.copy())
407 if child_hierarchy:
408 children.append({
409 "object": child_hierarchy,
410 "relationship": link.r_name
411 })
413 return {
414 "id": obj.id,
415 "name": obj.name,
416 "version": obj.version,
417 "type_id": str(obj.type_id),
418 "created_on": obj.created_on,
419 "created_by": obj.created_by,
420 "children": children,
421 "depth": current_depth
422 }
424 hierarchy = build_hierarchy_with_depth(object_id, 0)
425 return hierarchy