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):
        rec = SchemaService.get(db, id)
        if not obj:
            raise NotFoundError('Schema not exist')
        for k,v in data.dict(exclude_unset=True).items():
            setattr(rec, k, v)
        db.commit()
        db.refresh(rec)
        return rec

    def delete(db: Session, id:int):
        rec = SchemaService.get(db, id)
        if not obj:
            raise NotFoundError('Schema not exist')
        db.delete(rec)
        db.commit()