53 lines
1.7 KiB
Python
53 lines
1.7 KiB
Python
from typing import Type, TypeVar, Generic, List, Optional, Any
|
|
from sqlalchemy.orm import Session
|
|
from app.db.database import Base
|
|
from app.services.service_manager import ServiceManager
|
|
|
|
T = TypeVar('T')
|
|
|
|
class BaseService(Generic[T]):
|
|
def __init__(self, model: Type[T]):
|
|
self.model = model
|
|
self.service_manager = ServiceManager()
|
|
self._services = {}
|
|
|
|
def get(self, db: Session, id: int) -> Optional[T]:
|
|
return db.query(self.model).filter(self.model.id == id).first()
|
|
|
|
def get_all(self, db: Session, skip: int = 0, limit: int = 100) -> List[T]:
|
|
return db.query(self.model).offset(skip).limit(limit).all()
|
|
|
|
def create(self, db: Session, obj_in: dict) -> T:
|
|
db_obj = self.model(**obj_in)
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def update(self, db: Session, db_obj: T, obj_in: dict) -> T:
|
|
for field in obj_in:
|
|
if hasattr(db_obj, field):
|
|
setattr(db_obj, field, obj_in[field])
|
|
db.add(db_obj)
|
|
db.commit()
|
|
db.refresh(db_obj)
|
|
return db_obj
|
|
|
|
def delete(self, db: Session, id: int) -> bool:
|
|
obj = db.query(self.model).filter(self.model.id == id).first()
|
|
if obj:
|
|
db.delete(obj)
|
|
db.commit()
|
|
return True
|
|
return False
|
|
|
|
def get_service(self, name: str) -> Any:
|
|
"""Get a service by name with lazy loading"""
|
|
if name not in self._services:
|
|
self._services[name] = self.service_manager.get_service(name)
|
|
return self._services[name]
|
|
|
|
@property
|
|
def file_service(self):
|
|
"""Convenience property for file service"""
|
|
return self.get_service('file') |