webapp/app/services/configuration_service.py

from sqlalchemy.orm import Session
from typing import List, Optional
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.configuration import Configuration
from app.schemas.configuration import ConfigurationCreate, ConfigurationUpdate


class ConfigurationService:
    @staticmethod
    def create(db: Session, config_in: ConfigurationCreate) -> Configuration:
        config = db.query(Configuration).filter(Configuration.config_key == config_in.config_key).first()
        if config:
            raise BizLogicError(f"Config Key '{config_in.config_key}' already exist")
        db_obj = Configuration(**config_in.dict())
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    @staticmethod
    def get_all(db: Session) -> List[Configuration]:
        return db.query(Configuration).order_by(Configuration.id.desc()).all()

    @staticmethod
    def get_by_key(db: Session, config_key: str) -> Optional[Configuration]:
        config = db.query(Configuration).filter(Configuration.config_key == config_key).first()
        if not config:
            raise NotFoundError(f"Config Key '{config_key}' not exist")
        return config

    @staticmethod
    def update_by_key(db: Session, config_key: str, config_in: ConfigurationUpdate) -> Optional[Configuration]:
        db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
        if not db_obj:
            raise NotFoundError(f"Config Key '{config_key}' not exist")
        for field, value in config_in.dict(exclude_unset=True).items():
            setattr(db_obj, field, value)
        db.add(db_obj)
        db.commit()
        db.refresh(db_obj)
        return db_obj

    @staticmethod
    def delete_by_key(db: Session, config_key: str) -> Optional[Configuration]:
        db_obj = db.query(Configuration).filter(Configuration.config_key == config_key).first()
        if not db_obj:
            raise NotFoundError(f"Config Key '{config_key}' not exist")
        db.delete(db_obj)
        db.commit()
        return db_obj