Coverage for dormatory/api/routes/attributes.py: 93%
148 statements
« prev ^ index » next coverage.py v7.10.1, created at 2025-08-04 00:22 +0000
1"""
2Attributes API routes for DORMATORY.
4This module provides RESTful API endpoints for managing attribute 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 Attributes, Object
16router = APIRouter(tags=["attributes"])
19class AttributeCreate(BaseModel):
20 name: str
21 value: str
22 object_id: int
23 created_on: str
24 updated_on: str
27class AttributeUpdate(BaseModel):
28 name: Optional[str] = None
29 value: Optional[str] = None
30 updated_on: Optional[str] = None
33class AttributeResponse(BaseModel):
34 id: int
35 name: str
36 value: str
37 object_id: int
38 created_on: str
39 updated_on: str
41 class Config:
42 from_attributes = True
45@router.post("/", response_model=AttributeResponse)
46async def create_attribute(attribute_data: AttributeCreate, db: Session = Depends(get_db)):
47 """
48 Create a new attribute.
50 Args:
51 attribute_data: Attribute creation data
52 db: Database session
54 Returns:
55 Created attribute data
56 """
57 # Verify that the object exists
58 object_obj = db.query(Object).filter(Object.id == attribute_data.object_id).first()
59 if not object_obj:
60 raise HTTPException(status_code=404, detail="Object not found")
62 # Check if attribute already exists for this object
63 existing_attribute = db.query(Attributes).filter(
64 Attributes.object_id == attribute_data.object_id,
65 Attributes.name == attribute_data.name
66 ).first()
68 if existing_attribute:
69 raise HTTPException(status_code=409, detail="Attribute already exists for this object")
71 # Create the attribute
72 db_attribute = Attributes(
73 name=attribute_data.name,
74 value=attribute_data.value,
75 object_id=attribute_data.object_id,
76 created_on=attribute_data.created_on,
77 updated_on=attribute_data.updated_on
78 )
80 db.add(db_attribute)
81 db.commit()
82 db.refresh(db_attribute)
84 return AttributeResponse.from_orm(db_attribute)
87@router.get("/{attribute_id}", response_model=AttributeResponse)
88async def get_attribute_by_id(attribute_id: int, db: Session = Depends(get_db)):
89 """
90 Get an attribute by its ID.
92 Args:
93 attribute_id: Attribute ID
94 db: Database session
96 Returns:
97 Attribute data
98 """
99 db_attribute = db.query(Attributes).filter(Attributes.id == attribute_id).first()
100 if not db_attribute:
101 raise HTTPException(status_code=404, detail="Attribute not found")
103 return AttributeResponse.from_orm(db_attribute)
106@router.get("/", response_model=List[AttributeResponse])
107async def get_all_attributes(
108 skip: int = 0,
109 limit: int = 100,
110 object_id: Optional[int] = None,
111 name: Optional[str] = None,
112 db: Session = Depends(get_db)
113):
114 """
115 Get all attributes with optional filtering.
117 Args:
118 skip: Number of records to skip
119 limit: Maximum number of records to return
120 object_id: Filter by object ID
121 name: Filter by attribute name
122 db: Database session
124 Returns:
125 List of attributes
126 """
127 query = db.query(Attributes)
129 # Apply filters
130 if object_id:
131 query = query.filter(Attributes.object_id == object_id)
132 if name:
133 query = query.filter(Attributes.name == name)
135 # Apply pagination
136 attributes = query.offset(skip).limit(limit).all()
138 return [AttributeResponse.from_orm(attr) for attr in attributes]
141@router.put("/{attribute_id}", response_model=AttributeResponse)
142async def update_attribute(attribute_id: int, attribute_data: AttributeUpdate, db: Session = Depends(get_db)):
143 """
144 Update an existing attribute.
146 Args:
147 attribute_id: Attribute ID to update
148 attribute_data: Updated attribute data
149 db: Database session
151 Returns:
152 Updated attribute data
153 """
154 db_attribute = db.query(Attributes).filter(Attributes.id == attribute_id).first()
155 if not db_attribute:
156 raise HTTPException(status_code=404, detail="Attribute not found")
158 # Update fields if provided
159 if attribute_data.name is not None:
160 # Check if the new name conflicts with existing attribute for the same object
161 if attribute_data.name != db_attribute.name:
162 existing_attribute = db.query(Attributes).filter(
163 Attributes.object_id == db_attribute.object_id,
164 Attributes.name == attribute_data.name,
165 Attributes.id != attribute_id
166 ).first()
168 if existing_attribute:
169 raise HTTPException(status_code=409, detail="Attribute name already exists for this object")
171 db_attribute.name = attribute_data.name
173 if attribute_data.value is not None:
174 db_attribute.value = attribute_data.value
176 if attribute_data.updated_on is not None:
177 db_attribute.updated_on = attribute_data.updated_on
179 db.commit()
180 db.refresh(db_attribute)
182 return AttributeResponse.from_orm(db_attribute)
185@router.delete("/{attribute_id}")
186async def delete_attribute(attribute_id: int, db: Session = Depends(get_db)):
187 """
188 Delete an attribute.
190 Args:
191 attribute_id: Attribute ID to delete
192 db: Database session
194 Returns:
195 Success message
196 """
197 db_attribute = db.query(Attributes).filter(Attributes.id == attribute_id).first()
198 if not db_attribute:
199 raise HTTPException(status_code=404, detail="Attribute not found")
201 db.delete(db_attribute)
202 db.commit()
204 return {"message": "Attribute deleted successfully"}
207@router.post("/bulk", response_model=List[AttributeResponse])
208async def create_attributes_bulk(attribute_data: List[AttributeCreate], db: Session = Depends(get_db)):
209 """
210 Create multiple attributes in a single operation.
212 Args:
213 attribute_data: List of attribute creation data
214 db: Database session
216 Returns:
217 List of created attributes
218 """
219 created_attributes = []
221 for item in attribute_data:
222 # Verify that the object exists
223 object_obj = db.query(Object).filter(Object.id == item.object_id).first()
224 if not object_obj:
225 raise HTTPException(status_code=404, detail=f"Object {item.object_id} not found")
227 # Check if attribute already exists for this object
228 existing_attribute = db.query(Attributes).filter(
229 Attributes.object_id == item.object_id,
230 Attributes.name == item.name
231 ).first()
233 if existing_attribute:
234 raise HTTPException(status_code=409, detail=f"Attribute {item.name} already exists for object {item.object_id}")
236 # Create the attribute
237 db_attribute = Attributes(
238 name=item.name,
239 value=item.value,
240 object_id=item.object_id,
241 created_on=item.created_on,
242 updated_on=item.updated_on
243 )
245 db.add(db_attribute)
246 created_attributes.append(db_attribute)
248 db.commit()
250 # Refresh all attributes to get their IDs
251 for attr in created_attributes:
252 db.refresh(attr)
254 return [AttributeResponse.from_orm(attr) for attr in created_attributes]
257@router.get("/object/{object_id}")
258async def get_attributes_by_object(object_id: int, db: Session = Depends(get_db)):
259 """
260 Get all attributes for a specific object.
262 Args:
263 object_id: Object ID
264 db: Database session
266 Returns:
267 List of attributes for the object
268 """
269 # Verify that the object exists
270 object_obj = db.query(Object).filter(Object.id == object_id).first()
271 if not object_obj:
272 raise HTTPException(status_code=404, detail="Object not found")
274 attributes = db.query(Attributes).filter(Attributes.object_id == object_id).all()
276 return [AttributeResponse.from_orm(attr) for attr in attributes]
279@router.get("/name/{name}")
280async def get_attribute_by_name(name: str, db: Session = Depends(get_db)):
281 """
282 Get all attributes with a specific name.
284 Args:
285 name: Attribute name
286 db: Database session
288 Returns:
289 List of attributes with this name
290 """
291 attributes = db.query(Attributes).filter(Attributes.name == name).all()
293 return [AttributeResponse.from_orm(attr) for attr in attributes]
296@router.get("/object/{object_id}/map")
297async def get_object_attributes_map(object_id: int, db: Session = Depends(get_db)):
298 """
299 Get all attributes for an object as a key-value map.
301 Args:
302 object_id: Object ID
303 db: Database session
305 Returns:
306 Dictionary of attribute name-value pairs
307 """
308 # Verify that the object exists
309 object_obj = db.query(Object).filter(Object.id == object_id).first()
310 if not object_obj:
311 raise HTTPException(status_code=404, detail="Object not found")
313 attributes = db.query(Attributes).filter(Attributes.object_id == object_id).all()
315 return {attr.name: attr.value for attr in attributes}
318@router.post("/object/{object_id}/set")
319async def set_object_attributes(object_id: int, attributes: dict, db: Session = Depends(get_db)):
320 """
321 Set multiple attributes for an object.
323 Args:
324 object_id: Object ID
325 attributes: Dictionary of attribute name-value pairs
326 db: Database session
328 Returns:
329 Success message
330 """
331 # Verify that the object exists
332 object_obj = db.query(Object).filter(Object.id == object_id).first()
333 if not object_obj:
334 raise HTTPException(status_code=404, detail="Object not found")
336 from datetime import datetime, UTC
338 current_time = datetime.now(UTC).isoformat()
339 created_attributes = []
341 for name, value in attributes.items():
342 # Check if attribute already exists
343 existing_attribute = db.query(Attributes).filter(
344 Attributes.object_id == object_id,
345 Attributes.name == name
346 ).first()
348 if existing_attribute:
349 # Update existing attribute
350 existing_attribute.value = str(value)
351 existing_attribute.updated_on = current_time
352 else:
353 # Create new attribute
354 db_attribute = Attributes(
355 name=name,
356 value=str(value),
357 object_id=object_id,
358 created_on=current_time,
359 updated_on=current_time
360 )
361 db.add(db_attribute)
362 created_attributes.append(db_attribute)
364 db.commit()
366 return {"message": f"Set {len(attributes)} attributes for object {object_id}"}
369@router.delete("/object/{object_id}/name/{name}")
370async def delete_attribute_by_name(object_id: int, name: str, db: Session = Depends(get_db)):
371 """
372 Delete a specific attribute by name for an object.
374 Args:
375 object_id: Object ID
376 name: Attribute name
377 db: Database session
379 Returns:
380 Success message
381 """
382 # Verify that the object exists
383 object_obj = db.query(Object).filter(Object.id == object_id).first()
384 if not object_obj:
385 raise HTTPException(status_code=404, detail="Object not found")
387 # Find and delete the attribute
388 db_attribute = db.query(Attributes).filter(
389 Attributes.object_id == object_id,
390 Attributes.name == name
391 ).first()
393 if not db_attribute:
394 raise HTTPException(status_code=404, detail="Attribute not found")
396 db.delete(db_attribute)
397 db.commit()
399 return {"message": f"Attribute '{name}' deleted for object {object_id}"}
402@router.get("/search/{query}")
403async def search_attributes(query: str, db: Session = Depends(get_db)):
404 """
405 Search attributes by name or value.
407 Args:
408 query: Search query
409 db: Database session
411 Returns:
412 List of matching attributes
413 """
414 # Search in both name and value fields
415 attributes = db.query(Attributes).filter(
416 (Attributes.name.contains(query)) | (Attributes.value.contains(query))
417 ).all()
419 return [AttributeResponse.from_orm(attr) for attr in attributes]