from sqlalchemy.orm import Session 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]: return db.query(HttpSession).filter(HttpSession.session_id == session_id).first() @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: return False db.delete(obj) db.commit() return True @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: return None for k, v in data.dict().items(): if v is not None: setattr(obj, k, v) db.commit() db.refresh(obj) return obj