tegwick
ba535e1f8f
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a028f0-a42f-7582-89a8-ebaad7343834
128 lines
4.1 KiB
Python
128 lines
4.1 KiB
Python
"""Portable SQLAlchemy schema shared by the runtime store and Alembic."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import (
|
|
JSON,
|
|
Boolean,
|
|
Column,
|
|
DateTime,
|
|
ForeignKey,
|
|
Index,
|
|
Integer,
|
|
MetaData,
|
|
String,
|
|
Table,
|
|
Text,
|
|
create_engine,
|
|
event,
|
|
)
|
|
from sqlalchemy.engine import Engine
|
|
|
|
NAMING_CONVENTION = {
|
|
"ix": "ix_%(table_name)s_%(column_0_name)s",
|
|
"uq": "uq_%(table_name)s_%(column_0_name)s",
|
|
"ck": "ck_%(table_name)s_%(constraint_name)s",
|
|
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
|
|
"pk": "pk_%(table_name)s",
|
|
}
|
|
|
|
metadata = MetaData(naming_convention=NAMING_CONVENTION)
|
|
|
|
repositories = Table(
|
|
"repositories",
|
|
metadata,
|
|
Column("id", String(36), primary_key=True),
|
|
Column("slug", String(200), nullable=False, unique=True),
|
|
Column("checkout_path", Text, nullable=True),
|
|
Column("active", Boolean, nullable=False),
|
|
Column("last_attempt_at", DateTime(timezone=True), nullable=True),
|
|
Column("last_success_at", DateTime(timezone=True), nullable=True),
|
|
Column("last_status", String(50), nullable=True),
|
|
Column("last_source", String(200), nullable=True),
|
|
Column("created_at", DateTime(timezone=True), nullable=False),
|
|
Column("updated_at", DateTime(timezone=True), nullable=False),
|
|
)
|
|
|
|
snapshots = Table(
|
|
"snapshots",
|
|
metadata,
|
|
Column("id", String(36), primary_key=True),
|
|
Column(
|
|
"repo_id",
|
|
String(36),
|
|
ForeignKey("repositories.id", ondelete="RESTRICT"),
|
|
nullable=False,
|
|
),
|
|
Column("snapshot_at", DateTime(timezone=True), nullable=False),
|
|
Column("source", String(200), nullable=False),
|
|
Column("status", String(50), nullable=False),
|
|
Column("entry_count", Integer, nullable=False),
|
|
Column("source_revision", String(200), nullable=True),
|
|
Column("sources_json", JSON, nullable=False),
|
|
Column("errors_json", JSON, nullable=False),
|
|
Column("legacy_id", String(100), nullable=True, unique=True),
|
|
Column("created_at", DateTime(timezone=True), nullable=False),
|
|
)
|
|
|
|
entries = Table(
|
|
"entries",
|
|
metadata,
|
|
Column("id", String(36), primary_key=True),
|
|
Column(
|
|
"repo_id",
|
|
String(36),
|
|
ForeignKey("repositories.id", ondelete="RESTRICT"),
|
|
nullable=False,
|
|
),
|
|
Column(
|
|
"snapshot_id",
|
|
String(36),
|
|
ForeignKey("snapshots.id", ondelete="RESTRICT"),
|
|
nullable=False,
|
|
),
|
|
Column("package_name", String(300), nullable=False),
|
|
Column("package_version", String(100), nullable=True),
|
|
Column("ecosystem", String(50), nullable=False),
|
|
Column("license_spdx", String(100), nullable=True),
|
|
Column("is_direct", Boolean, nullable=False),
|
|
Column("is_dev", Boolean, nullable=False),
|
|
Column("source_path", Text, nullable=True),
|
|
Column("created_at", DateTime(timezone=True), nullable=False),
|
|
)
|
|
|
|
Index("ix_snapshots_repo_time", snapshots.c.repo_id, snapshots.c.snapshot_at)
|
|
Index("ix_entries_snapshot", entries.c.snapshot_id)
|
|
Index("ix_entries_repo", entries.c.repo_id)
|
|
Index("ix_entries_license", entries.c.license_spdx)
|
|
|
|
|
|
def database_url(value: str | Path) -> str:
|
|
"""Normalize a filesystem path or supported URL into a SQLAlchemy URL."""
|
|
text = str(value)
|
|
if "://" not in text:
|
|
return f"sqlite:///{Path(text).resolve()}"
|
|
if text.startswith("postgres://"):
|
|
return text.replace("postgres://", "postgresql+psycopg://", 1)
|
|
if text.startswith("postgresql://"):
|
|
return text.replace("postgresql://", "postgresql+psycopg://", 1)
|
|
return text
|
|
|
|
|
|
def create_database_engine(value: str | Path) -> Engine:
|
|
url = database_url(value)
|
|
options: dict[str, object] = {"pool_pre_ping": True}
|
|
if url.startswith("sqlite:"):
|
|
options["connect_args"] = {"check_same_thread": False}
|
|
engine = create_engine(url, **options)
|
|
if engine.dialect.name == "sqlite":
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def _enable_sqlite_foreign_keys(dbapi_connection, _connection_record) -> None:
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.execute("PRAGMA foreign_keys = ON")
|
|
cursor.close()
|
|
|
|
return engine
|