feat(release): add verified image publication pipeline
This commit is contained in:
+22
-9
@@ -2,19 +2,33 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from importlib import resources
|
||||
|
||||
import psycopg
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
import psycopg
|
||||
|
||||
from app.core.config import Settings
|
||||
|
||||
|
||||
def migration_root():
|
||||
"""Return the packaged Alembic resource tree for source and wheel installs."""
|
||||
return resources.files("app.db.alembic")
|
||||
|
||||
|
||||
def migration_inventory() -> tuple[str, ...]:
|
||||
"""Return the deterministic, non-empty migration revision inventory."""
|
||||
versions = migration_root().joinpath("versions")
|
||||
inventory = tuple(sorted(item.name for item in versions.iterdir() if item.name.endswith(".py") and item.name != "__init__.py"))
|
||||
if not inventory:
|
||||
raise RuntimeError("packaged Alembic migration inventory is empty")
|
||||
return inventory
|
||||
|
||||
|
||||
def get_alembic_config(settings: Settings) -> Config:
|
||||
project_root = Path(__file__).resolve().parents[2]
|
||||
config = Config(str(project_root / "alembic.ini"))
|
||||
config.set_main_option("script_location", str(project_root / "migrations"))
|
||||
root = migration_root()
|
||||
config = Config(str(root.joinpath("alembic.ini")))
|
||||
config.set_main_option("script_location", str(root))
|
||||
config.set_main_option("sqlalchemy.url", settings.active_database_url.replace("%", "%%"))
|
||||
# migrations/env.py must use this explicit settings object instead of the
|
||||
# process-global cached Settings. This keeps programmatic callers (notably
|
||||
@@ -73,10 +87,9 @@ def _connect(settings: Settings) -> psycopg.Connection:
|
||||
|
||||
|
||||
def _has_alembic_version_table(settings: Settings) -> bool:
|
||||
with _connect(settings) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT to_regclass('public.alembic_version')")
|
||||
return cur.fetchone()[0] is not None
|
||||
with _connect(settings) as conn, conn.cursor() as cur:
|
||||
cur.execute("SELECT to_regclass('public.alembic_version')")
|
||||
return cur.fetchone()[0] is not None
|
||||
|
||||
|
||||
def _has_legacy_portfolio_schema(settings: Settings) -> bool:
|
||||
|
||||
Reference in New Issue
Block a user