Database Layer

db.models

SQLAlchemy ORM models for Lingua Loop.

class lingua_loop.db.models.Segment(**kwargs)

Bases: Base

Represents a segment within a transcript.

duration: Mapped[float]

id: Mapped[int]

start: Mapped[float]

text: Mapped[str]

transcript: Mapped[Transcript]

transcript_id: Mapped[str]

class lingua_loop.db.models.Transcript(**kwargs)

Bases: Base

One transcript per video for now.

created_at: Mapped[DateTime]

is_generated: Mapped[bool]

language_code: Mapped[SupportedLanguageCodes]

segments: Mapped[List[Segment]]

video_id: Mapped[str]

db.session

Database session management utilities.

async lingua_loop.db.session.create_db_and_tables(async_engine: AsyncEngine)

Create all database tables defined in models.

async lingua_loop.db.session.get_async_session(request: Request) → AsyncGenerator[AsyncSession, None]

Provide an async database session for dependency injection.

lingua_loop.db.session.get_engine_and_session_maker(db_driver: str = 'sqlite+aiosqlite', database_path: Path | str = PosixPath('/home/runner/.lingua_loop/transcripts.db')) → Tuple[AsyncEngine, async_sessionmaker[AsyncSession]]

Create and return the async engine and session maker.

async lingua_loop.db.session.shutdown(async_engine: AsyncEngine)

Dispose of the async engine.

db.transcript

CRUD operations for transcripts.

async lingua_loop.db.transcript.read_or_create_transcript_with_segments(video_id: str, language_code: SupportedLanguageCodes, session: AsyncSession) → Transcript

Get or create a transcript with all segments for the given video.