Skip to content

aria2_server.db.migrations

__all__ module-attribute 

__all__ = (
    "alembic_ini",
    "get_default_cfg",
    "revision",
    "run_async_upgrade",
    "script_location",
    "upgrade",
)

alembic_ini module-attribute 

alembic_ini = _here / 'alembic.ini'

script_location module-attribute 

script_location = _here / '_alembic'

upgrade module-attribute 

upgrade = partial(
    upgrade, config=get_default_cfg(), revision="head"
)

Note: this function is designed to be used in cli.

revision module-attribute 

revision = partial(
    revision, config=get_default_cfg(), autogenerate=True
)

Note: this function is designed to be used in cli.

get_default_cfg 

get_default_cfg() -> Config
Source code in src/aria2_server/db/migrations/__init__.py 
28
29
30
31
32
33
def get_default_cfg() -> config.Config:
    cfg = config.Config(alembic_ini)
    cfg.set_main_option("script_location", str(script_location))
    cfg.set_main_option("sqlalchemy.url", str(DATABASE_URL))
    cfg.attributes["target_metadata"] = Base.metadata
    return cfg

run_async_upgrade async 

run_async_upgrade() -> None

This function is designed to be used as a library api.

The difference is that this function will not configure logging settings.

Source code in src/aria2_server/db/migrations/__init__.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def run_async_upgrade() -> None:
    """This function is designed to be used as a library api.

    The difference is that this function will not configure logging settings.
    """

    # Refer: https://alembic.sqlalchemy.org/en/latest/cookbook.html#programmatic-api-use-connection-sharing-with-asyncio
    def run_upgrade(connection: Connection, cfg: config.Config) -> None:
        cfg.attributes["connection"] = connection
        # NOTE: `no_logging_config` is a custom option used by `aria2-server`,
        # see `env.py` for more details.
        cfg.set_main_option("no_logging_config", "true")
        command.upgrade(cfg, "head")

    async with engine.begin() as conn:
        await conn.run_sync(run_upgrade, get_default_cfg())