Spaces:
Runtime error
Runtime error
File size: 1,686 Bytes
3dcada4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | from sqlalchemy import Column, String, TIMESTAMP
from sqlalchemy.sql import func
from .database import Base
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import relationship
class Session(Base):
__tablename__ = "sessions"
session_id = Column(String, primary_key=True, index=True)
created_at = Column(TIMESTAMP(timezone=True), server_default=func.now())
status = Column(String, default="active")
transcriptions = relationship(
"Transcription",
back_populates="session",
cascade="all, delete-orphan",
passive_deletes=True
)
# CRUD operations
async def create_session(db: AsyncSession, session_id: str, status: str = "active"):
new_session = Session(session_id=session_id, status=status)
db.add(new_session)
await db.commit()
await db.refresh(new_session)
return new_session
async def get_session(db: AsyncSession, session_id: str):
result = await db.execute(select(Session).where(Session.session_id == session_id))
return result.scalar_one_or_none()
async def get_all_sessions(db: AsyncSession):
result = await db.execute(select(Session))
return result.scalars().all()
async def update_session_status(db: AsyncSession, session_id: str, status: str):
session = await get_session(db, session_id)
if session:
session.status = status
await db.commit()
await db.refresh(session)
return session
async def delete_session(db: AsyncSession, session_id: str):
session = await get_session(db, session_id)
if session:
await db.delete(session)
await db.commit()
return session
|