agilentics / boiler 
 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
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
"""Alembic migration environment.

The connection URL comes from the same place the app's does (utils.db_wrapper),
so there is one definition of how a connection string is made instead of two
that can drift.

Local dev / CI: TCP to localhost:5432 (Cloud SQL Auth Proxy or local Postgres)
Cloud SQL:      Unix socket, when INSTANCE_UNIX_SOCKET is set
"""
from logging.config import fileConfig

from alembic import context
from sqlalchemy import pool

from db.db_declarations import Base
from utils.db_wrapper import db_factory

config = context.config
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata


def include_object(obj, name, type_, reflected, compare_to) -> bool:
    """Ignore tables this application does not declare.

    A developer's local Postgres may carry another project's tables under the
    same role, and autogenerate would otherwise propose dropping them - and make
    `alembic check` fail for a reason unrelated to this codebase.

    The cost is that removing a model no longer autogenerates its DROP, which is
    an acceptable trade: dropping a table should be written on purpose, not be a
    side effect of deleting a class.
    """
    if type_ == "table" and reflected and name not in target_metadata.tables:
        return False
    return True


_OPTIONS = {
    "target_metadata": target_metadata,
    "include_object": include_object,
    # Without this, widening a String or changing nullability renders as no
    # change at all, and the schema silently drifts from the models.
    "compare_type": True,
}


def run_migrations_offline() -> None:
    context.configure(
        url=str(db_factory().url),
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        **_OPTIONS,
    )
    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    connectable = db_factory().execution_options(poolclass=pool.NullPool)
    with connectable.connect() as connection:
        context.configure(connection=connection, **_OPTIONS)
        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()