webapp/app/services/http_session_service.py

from sqlalchemy.orm import Session
from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
from app.models.http_session import HttpSession
from app.schemas.http_session import HttpSessionCreate, HttpSessionUpdate
from typing import Optional

class HttpSessionService:

    @staticmethod
    def create(db: Session, data: HttpSessionCreate) -> HttpSession:
        obj = HttpSession(**data.dict())
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj

    @staticmethod
    def get_by_sid(db: Session, session_id: str) -> Optional[HttpSession]:
        obj = db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
        if not obj:
            raise NotFoundError("Session not found")
        return obj

    @staticmethod
    def delete_by_sid(db: Session, session_id: str) -> bool:
        obj = db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
        if not obj:
            raise NotFoundError("Session not found")
        db.delete(obj)
        db.commit()
        return obj

    @staticmethod
    def update_by_sid(db: Session, session_id: str, data: HttpSessionUpdate):
        obj = db.query(HttpSession).filter(HttpSession.session_id == session_id).first()
        if not obj:
            raise NotFoundError("Session not found")
        for k, v in data.dict().items():
            if v is not None:
                setattr(obj, k, v)
        db.commit()
        db.refresh(obj)
        return obj