webapp/app/services/schema_service.py

# app/services/schema_service.py
from sqlalchemy.orm import Session
from typing import Optional
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.schema import VasSchema
from app.schemas.schema import VasSchemaCreate, VasSchemaUpdate

class SchemaService:

    def create(db: Session, data: VasSchemaCreate):
        rec = VasSchema(**data.dict())
        db.add(rec)
        db.commit()
        db.refresh(rec)
        return rec

    def get(db: Session, id: int):
        obj = db.query(VasSchema).filter_by(id=id).first()
        if not obj:
            raise NotFoundError('Schema not exist')
        return obj

    def update(db: Session, id: int, data: VasSchemaUpdate):
        obj = db.query(VasSchema).filter_by(id=id).first()
        if not obj:
            raise NotFoundError('Schema not exist')
        for k,v in data.dict(exclude_unset=True).items():
            setattr(obj, k, v)
        db.commit()
        db.refresh(obj)
        return obj

    def delete(db: Session, id:int):
        obj = db.query(VasSchema).filter_by(id=id).first()
        if not obj:
            raise NotFoundError('Schema not exist')
        db.delete(obj)
        db.commit()

    def list_all(db: Session):
        return db.query(VasSchema).all()