import logging from datetime import datetime from typing import Any, Dict, Generic, List, Optional, Tuple, Type, TypeVar, Union from fastapi.encoders import jsonable_encoder from pydantic import BaseModel from sqlalchemy.exc import IntegrityError, NoResultFound from sqlalchemy.orm import Session from src.utils.enums import DBSortTypes from src.database.base import Base logger = logging.getLogger(__name__) ModelType = TypeVar("ModelType", bound=Base) CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel) UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel) DetailSchemaType = TypeVar("DetailSchemaType", bound=BaseModel) class BaseRepository( Generic[ModelType, CreateSchemaType, UpdateSchemaType, DetailSchemaType] ): def __init__( self, model: Type[ModelType], schema: Type[DetailSchemaType], use_soft_delete: bool = False, ): """Repository object with default methods to Create, Read, Update, Delete (CRUD). Parameters: - db: SQLAlchemy Session - model: Database Model (A SQLAlchemy model class) - schema: DTO / Data Transfer Object (A Pydantic model (schema) class) - use_soft_delete: If set to true, model will exclude soft deleted records """ self.model = model self.schema = schema self.use_soft_delete = use_soft_delete def get(self, db: Session, _id: Any) -> Optional[Tuple[DetailSchemaType, str]]: try: query = db.query(self.model) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) result = query.filter(self.model.id == _id).first() if not result: msg = "No such resource found" return (None, msg) return (self.schema.model_validate(result), "Resource details") except Exception as e: msg = str(e) logger.error(msg) return (None, msg) def get_raw(self, db: Session, _id: int) -> Optional[ModelType]: query = db.query(self.model).filter(self.model.id == _id) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) return query.first() def get_by_name( self, db: Session, *, name_in: str ) -> Optional[Tuple[DetailSchemaType, str]]: try: query = db.query(self.model) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) result = query.filter(self.model.name == name_in).first() if not result: msg = "No such resource found" return (None, msg) return (self.schema.model_validate(result), "Resource details") except NoResultFound: msg = "No such resource found" logger.error(msg) return (None, msg) except Exception as e: msg = str(e) logger.error(msg) return (None, msg) def get_one(self, db: Session, **filters) -> Optional[ModelType]: query = db.query(self.model).filter_by(**filters) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) return query.first() def get_many( self, db: Session, skip: int, limit: int, sort_by: str = "", sort_type: str = DBSortTypes.ASCENDING, **filters ) -> Optional[List[ModelType]]: sort_query = None if sort_by: sort_query = getattr(self.model, sort_by) if sort_query and sort_type == DBSortTypes.DESCENDING: sort_query = sort_query.desc() query = db.query(self.model).filter_by(**filters) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) result = query.order_by(sort_query).offset(skip).limit(limit).all() return result def get_ids_in(self, db: Session, ids_in: List[int]) -> Optional[List[ModelType]]: query = db.query(self.model).filter(self.model.id.in_(ids_in)) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) return query.all() def get_multi( self, db: Session, skip: int = 0, limit: Optional[int] = None ) -> Tuple[List[DetailSchemaType], str]: query = db.query(self.model).offset(skip).limit(limit) if self.use_soft_delete: query = query.filter(self.model.deleted_at.is_(None)) results = query.all() return [self.schema.model_validate(item) for item in results], "Resource List" def create( self, db: Session, *, obj_in: CreateSchemaType ) -> Optional[Tuple[ModelType, str]]: try: obj_in_data = jsonable_encoder(obj_in) result = self.model(**obj_in_data) # type: ignore db.add(result) db.commit() db.refresh(result) return (result, "Resource created") except IntegrityError: msg = "Attempting to create duplicate resource" logger.error(msg) return (None, msg) except Exception as e: msg = str(e) logger.error(msg) return (None, msg) def update( self, db: Session, *, result: ModelType, obj_in: Union[UpdateSchemaType, Dict[str, Any]], return_raw: bool = False ) -> DetailSchemaType: try: obj_data = jsonable_encoder(result) except Exception: obj_data = result.__dict__.copy() update_data = ( obj_in.model_dump(exclude_unset=True) if isinstance(obj_in, BaseModel) else obj_in ) for field in obj_data: if field in update_data: setattr(result, field, update_data[field]) db.add(result) db.commit() db.refresh(result) if return_raw: return result return self.schema.model_validate(result) def update_by_id( self, db: Session, object_id: int, payload: Union[UpdateSchemaType, Dict[str, Any]], return_raw: bool = False, ) -> Optional[Union[DetailSchemaType, ModelType]]: try: result = db.query(self.model).filter(self.model.id == object_id).first() except Exception: return None return self.update(result=result, obj_in=payload, return_raw=return_raw) def update_by_name( self, db: Session, name_in_db: str, obj_in: Union[UpdateSchemaType, Dict[str, Any]], ) -> Optional[ModelType]: try: result = db.query(self.model).filter(self.model.name == name_in_db).first() except Exception: return None return self.update(result=result, obj_in=obj_in, return_raw=True) def remove(self, db: Session, _id: int) -> Optional[DetailSchemaType]: result = db.query(self.model).get(_id) if result: db.delete(result) db.commit() return self.schema.model_validate(result) return None def soft_delete(self, db: Session, _id: int) -> Optional[Dict[str, bool]]: if not hasattr(self.model, "deleted_at"): return { "deleted": False, "message": "Model does not support soft delete. 'deleted_at' column was not found.", } result = db.query(self.model).get(_id) if result: result.deleted_at = datetime.now(datetime.utc) db.commit() return {"deleted": True} return {"deleted": False} def remove_by_param( self, db: Session, **params ) -> Optional[Tuple[DetailSchemaType, str]]: try: result = db.query(self.model).filter_by(**params).first() if result is None: raise Exception("No such resource found") db.delete(result) db.commit() return self.schema.model_validate(result), "Deleted successfully" except Exception as e: return None, str(e)