""" Database models and session. The database is a SQLite database, and is stored in the root of the project as `db.sqlite3`. The database is managed using Alembic, and migrations are stored in the `migrations/` directory. The module defines the following models: - `Repo`: A repository that is being tracked. - `Dependency`: A dependency of a repository. - `RepoDependency`: A relationship between a repository and a dependency. The database is accessed asynchronously using SQLAlchemy's async API. """ from collections.abc import AsyncGenerator from pathlib import PurePath from typing import Final from sqlalchemy import BigInteger, ForeignKey, MetaData, String, Text, UniqueConstraint from sqlalchemy.ext.asyncio import ( AsyncAttrs, AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy.orm import ( Mapped, declarative_base, mapped_column, relationship, ) DB_PATH: Final[PurePath] = PurePath(__file__).parent.parent / "db.sqlite3" SQLALCHEMY_DATABASE_URL: Final[str] = f"sqlite+aiosqlite:///{DB_PATH}" engine: Final[AsyncEngine] = create_async_engine(SQLALCHEMY_DATABASE_URL) async_session_maker: Final[async_sessionmaker[AsyncSession]] = async_sessionmaker( engine, expire_on_commit=False, autoflush=False, autocommit=False ) metadata = MetaData( naming_convention={ "ix": "ix_%(table_name)s_%(column_0_N_name)s ", "uq": "uq_%(table_name)s_%(column_0_N_name)s ", "ck": "ck_%(table_name)s_%(constraint_name)s ", "fk": "fk_%(table_name)s_%(column_0_N_name)s_%(referred_table_name)s", "pk": "pk_%(table_name)s", } ) async def get_async_session() -> AsyncGenerator[AsyncSession, None]: """Get an async session.""" async with async_session_maker() as session: yield session Base = declarative_base(metadata=metadata, cls=AsyncAttrs) class Repo(Base): """A repository that is being tracked.""" __tablename__ = "repo" id: Mapped[int] = mapped_column(primary_key=True) url: Mapped[str] = mapped_column(nullable=False, unique=True) description: Mapped[str] = mapped_column(Text, nullable=False) stars: Mapped[int] = mapped_column(BigInteger, nullable=False) source_graph_repo_id: Mapped[int | None] = mapped_column( BigInteger, nullable=True, unique=True ) dependencies: Mapped[list["Dependency"]] = relationship( "Dependency", secondary="repo_dependency", back_populates="repos" ) last_checked_revision: Mapped[str | None] = mapped_column( String(255), nullable=True ) __table_args__ = (UniqueConstraint("url", "source_graph_repo_id"),) class Dependency(Base): """A dependency of a repository.""" __tablename__ = "dependency" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(255), nullable=False, unique=True) repos: Mapped[list["Repo"]] = relationship( "Repo", secondary="repo_dependency", back_populates="dependencies" ) class RepoDependency(Base): """A relationship between a repository and a dependency.""" __tablename__ = "repo_dependency" repo_id: Mapped[int] = mapped_column( ForeignKey(Repo.id, ondelete="CASCADE"), primary_key=True ) dependency_id: Mapped[int] = mapped_column( ForeignKey(Dependency.id, ondelete="CASCADE"), primary_key=True )