configuration_service.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. from sqlalchemy.orm import Session
  2. from typing import List, Optional
  3. from app.models.configuration import Configuration
  4. from app.schemas.configuration import ConfigurationCreate, ConfigurationUpdate
  5. class ConfigurationService:
  6. @staticmethod
  7. def create(db: Session, config_in: ConfigurationCreate) -> Configuration:
  8. db_obj = Configuration(**config_in.dict())
  9. db.add(db_obj)
  10. db.commit()
  11. db.refresh(db_obj)
  12. return db_obj
  13. @staticmethod
  14. def get_all(db: Session) -> List[Configuration]:
  15. return db.query(Configuration).order_by(Configuration.id.desc()).all()
  16. @staticmethod
  17. def get_by_key(db: Session, config_key: str) -> Optional[Configuration]:
  18. return db.query(Configuration).filter(Configuration.config_key == config_key).first()
  19. @staticmethod
  20. def update_by_key(db: Session, config_key: str, config_in: ConfigurationUpdate) -> Optional[Configuration]:
  21. db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
  22. if not db_obj:
  23. return None
  24. for field, value in config_in.dict(exclude_unset=True).items():
  25. setattr(db_obj, field, value)
  26. db.add(db_obj)
  27. db.commit()
  28. db.refresh(db_obj)
  29. return db_obj
  30. @staticmethod
  31. def delete_by_key(db: Session, config_key: str) -> Optional[Configuration]:
  32. db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
  33. if not db_obj:
  34. return None
  35. db.delete(db_obj)
  36. db.commit()
  37. return db_obj