| 123456789101112131415161718192021222324252627282930313233343536373839 |
- from typing import Optional
- from sqlalchemy.ext.asyncio import AsyncSession
- from sqlalchemy import select, delete
- from app.core.biz_exception import NotFoundError
- from app.models.http_session import HttpSession
- from app.schemas.http_session import HttpSessionCreate
- class HttpSessionService:
- # ============================
- # 创建 Session
- # ============================
- @staticmethod
- async def create(db: AsyncSession, data: HttpSessionCreate) -> HttpSession:
- obj = HttpSession(**data.dict())
- db.add(obj)
- await db.commit()
- await db.refresh(obj)
- return obj
- # ============================
- # 根据 session_id 获取
- # ============================
- @staticmethod
- async def get_by_sid(
- db: AsyncSession,
- session_id: str
- ) -> HttpSession:
- stmt = select(HttpSession).where(HttpSession.session_id == session_id)
- result = await db.execute(stmt)
- obj = result.scalar_one_or_none()
- if not obj:
- raise NotFoundError("Session not found")
- return obj
|
