Felipe Cardoso
7d9f4ee4b2
Implemented user CRUD operations including creation, retrieval, updating, and deletion through a generic CRUD base class. Enhanced user schemas with additional attributes and created tests to verify functionality, covering edge cases such as duplicates and pagination. Updated the test suite with new fixtures and methods to support the tests.
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
from typing import Any, Dict, Generic, List, Optional, Type, TypeVar, Union
|
|
from fastapi.encoders import jsonable_encoder
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session
|
|
from app.core.database import Base
|
|
|
|
ModelType = TypeVar("ModelType", bound=Base)
|
|
CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
|
|
UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
|
|
|
|
|
|
class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
|
|
def __init__(self, model: Type[ModelType]):
|
|
"""
|
|
CRUD object with default methods to Create, Read, Update, Delete (CRUD).
|
|
|
|
Parameters:
|
|
model: A SQLAlchemy model class
|
|
"""
|
|
self.model = model
|
|
|
|
def get(self, db: Session, id: str) -> Optional[ModelType]:
|
|
return db.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_multi(
|
|
self, db: Session, *, skip: int = 0, limit: int = 100
|
|
) -> List[ModelType]:
|
|
return db.query(self.model).offset(skip).limit(limit).all()
|
|
|
|
def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
|
|
obj_in_data = jsonable_encoder(obj_in)
|
|
db_obj = self.model(**obj_in_data)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(
|
|
self,
|
|
db: Session,
|
|
*,
|
|
db_obj: ModelType,
|
|
obj_in: Union[UpdateSchemaType, Dict[str, Any]]
|
|
) -> ModelType:
|
|
obj_data = jsonable_encoder(db_obj)
|
|
if isinstance(obj_in, dict):
|
|
update_data = obj_in
|
|
else:
|
|
update_data = obj_in.model_dump(exclude_unset=True)
|
|
for field in obj_data:
|
|
if field in update_data:
|
|
setattr(db_obj, field, update_data[field])
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def remove(self, db: Session, *, id: str) -> ModelType:
|
|
obj = db.query(self.model).get(id)
|
|
db.delete(obj)
|
|
db.commit()
|
|
return obj |