|
| 1 | +from datetime import datetime |
1 | 2 | from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union |
2 | 3 |
|
3 | 4 | from fastapi.encoders import jsonable_encoder |
4 | 5 | from pydantic import BaseModel |
5 | 6 | from sqlalchemy.orm import Session |
6 | | -from datetime import datetime |
7 | 7 |
|
8 | 8 | from app.db.base_class import Base |
9 | 9 |
|
|
13 | 13 |
|
14 | 14 |
|
15 | 15 | class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]): |
16 | | - def __init__(self, *, model: Type[ModelType]): |
| 16 | + """ |
| 17 | + Generic CRUD operations for SQLAlchemy models. |
| 18 | +
|
| 19 | + Provides standard Create, Read, Update, Delete operations with type safety. |
| 20 | + """ |
| 21 | + |
| 22 | + def __init__(self, *, model: Type[ModelType]) -> None: |
17 | 23 | """ |
18 | | - CRUD object with default methods to Create, Read, Update, Delete (CRUD). |
19 | | - **Parameters** |
20 | | - * `model`: A SQLAlchemy model class |
21 | | - * `schema`: A Pydantic model (schema) class |
| 24 | + Initialize CRUD operations for a specific model. |
| 25 | +
|
| 26 | + Args: |
| 27 | + model: A SQLAlchemy model class |
22 | 28 | """ |
23 | 29 | self.model = model |
24 | 30 |
|
25 | 31 | def get(self, db: Session, id: str) -> Optional[ModelType]: |
| 32 | + """ |
| 33 | + Retrieve a single record by primary key. |
| 34 | +
|
| 35 | + Args: |
| 36 | + db: Database session |
| 37 | + id: Primary key value |
| 38 | +
|
| 39 | + Returns: |
| 40 | + Model instance or None if not found |
| 41 | + """ |
26 | 42 | return db.query(self.model).filter(self.model.id == id).first() |
27 | 43 |
|
28 | 44 | def get_by_field(self, db: Session, field: str, value: Any) -> Optional[ModelType]: |
| 45 | + """ |
| 46 | + Retrieve a single record by field value. |
| 47 | +
|
| 48 | + Args: |
| 49 | + db: Database session |
| 50 | + field: Field name to filter by |
| 51 | + value: Value to match |
| 52 | +
|
| 53 | + Returns: |
| 54 | + Model instance or None if not found |
| 55 | +
|
| 56 | + Raises: |
| 57 | + AttributeError: If field doesn't exist on model |
| 58 | + """ |
| 59 | + if not hasattr(self.model, field): |
| 60 | + raise AttributeError(f"Model {self.model.__name__} has no field '{field}'") |
| 61 | + |
29 | 62 | return db.query(self.model).filter(getattr(self.model, field) == value).first() |
30 | 63 |
|
31 | 64 | def get_multi_by_field( |
32 | | - self, db: Session, field: str, value: Any |
33 | | - ) -> Optional[ModelType]: |
34 | | - return db.query(self.model).filter(getattr(self.model, field) == value).all() |
| 65 | + self, db: Session, field: str, value: Any, *, skip: int = 0, limit: int = 100 |
| 66 | + ) -> List[ModelType]: |
| 67 | + """ |
| 68 | + Retrieve multiple records by field value. |
| 69 | +
|
| 70 | + Args: |
| 71 | + db: Database session |
| 72 | + field: Field name to filter by |
| 73 | + value: Value to match |
| 74 | + skip: Number of records to skip |
| 75 | + limit: Maximum number of records to return |
| 76 | +
|
| 77 | + Returns: |
| 78 | + List of model instances |
| 79 | +
|
| 80 | + Raises: |
| 81 | + AttributeError: If field doesn't exist on model |
| 82 | + """ |
| 83 | + if not hasattr(self.model, field): |
| 84 | + raise AttributeError(f"Model {self.model.__name__} has no field '{field}'") |
| 85 | + |
| 86 | + return ( |
| 87 | + db.query(self.model) |
| 88 | + .filter(getattr(self.model, field) == value) |
| 89 | + .offset(skip) |
| 90 | + .limit(limit) |
| 91 | + .all() |
| 92 | + ) |
35 | 93 |
|
36 | 94 | def get_multi( |
37 | 95 | self, db: Session, *, skip: int = 0, limit: int = 100 |
38 | 96 | ) -> List[ModelType]: |
| 97 | + """ |
| 98 | + Retrieve multiple records with pagination. |
| 99 | +
|
| 100 | + Args: |
| 101 | + db: Database session |
| 102 | + skip: Number of records to skip |
| 103 | + limit: Maximum number of records to return |
| 104 | +
|
| 105 | + Returns: |
| 106 | + List of model instances |
| 107 | + """ |
39 | 108 | return db.query(self.model).offset(skip).limit(limit).all() |
40 | 109 |
|
41 | 110 | def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType: |
| 111 | + """ |
| 112 | + Create a new record in the database. |
| 113 | +
|
| 114 | + Args: |
| 115 | + db: Database session |
| 116 | + obj_in: Pydantic schema with creation data |
| 117 | +
|
| 118 | + Returns: |
| 119 | + Created model instance |
| 120 | + """ |
42 | 121 | obj_in_data = jsonable_encoder(obj_in) |
43 | | - db_obj = self.model(**obj_in_data) # type: ignore |
| 122 | + db_obj = self.model(**obj_in_data) |
44 | 123 | db.add(db_obj) |
45 | 124 | db.commit() |
46 | 125 | db.refresh(db_obj) |
47 | 126 | return db_obj |
48 | 127 |
|
49 | 128 | def update( |
50 | | - self, db: Session, *, obj_in: Union[UpdateSchemaType, Dict[str, Any]] |
| 129 | + self, |
| 130 | + db: Session, |
| 131 | + *, |
| 132 | + db_obj: ModelType, |
| 133 | + obj_in: Union[UpdateSchemaType, Dict[str, Any]], |
51 | 134 | ) -> ModelType: |
52 | | - db_obj = db.query(self.model).get(obj_in.id) |
53 | | - obj_data = jsonable_encoder(db_obj) |
| 135 | + """ |
| 136 | + Update an existing record in the database. |
| 137 | +
|
| 138 | + Args: |
| 139 | + db: Database session |
| 140 | + db_obj: Database object to update |
| 141 | + obj_in: Pydantic schema or dict with updated data |
| 142 | +
|
| 143 | + Returns: |
| 144 | + Updated model instance |
| 145 | + """ |
| 146 | + # Convert input to dictionary |
54 | 147 | if isinstance(obj_in, dict): |
55 | 148 | update_data = obj_in |
56 | 149 | else: |
57 | | - update_data = jsonable_encoder(obj_in) |
58 | | - |
59 | | - # Manually set updated_at to current time |
60 | | - update_data['updated_at'] = datetime.utcnow() |
61 | | - |
62 | | - for field in obj_data: |
63 | | - if field in update_data: |
64 | | - setattr(db_obj, field, update_data[field]) |
| 150 | + update_data = jsonable_encoder(obj_in, exclude_unset=True) |
| 151 | + |
| 152 | + # Update fields that exist in input |
| 153 | + for field, value in update_data.items(): |
| 154 | + if hasattr(db_obj, field): |
| 155 | + setattr(db_obj, field, value) |
| 156 | + |
| 157 | + # Update timestamp if model has updated_at field |
| 158 | + if hasattr(db_obj, "updated_at"): |
| 159 | + db_obj.updated_at = datetime.utcnow() |
| 160 | + |
65 | 161 | db.add(db_obj) |
66 | 162 | db.commit() |
67 | 163 | db.refresh(db_obj) |
68 | 164 | return db_obj |
69 | 165 |
|
70 | 166 | def remove(self, db: Session, *, id: int) -> ModelType: |
71 | | - obj = db.query(self.model).get(id) |
| 167 | + """ |
| 168 | + Delete a record from the database. |
| 169 | +
|
| 170 | + Args: |
| 171 | + db: Database session |
| 172 | + id: Primary key of record to delete |
| 173 | +
|
| 174 | + Returns: |
| 175 | + Deleted model instance |
| 176 | +
|
| 177 | + Raises: |
| 178 | + ValueError: If record not found |
| 179 | + """ |
| 180 | + obj = db.query(self.model).filter(self.model.id == id).first() |
| 181 | + if not obj: |
| 182 | + raise ValueError(f"Record with id {id} not found") |
| 183 | + |
72 | 184 | db.delete(obj) |
73 | 185 | db.commit() |
74 | 186 | return obj |
0 commit comments