upgrade database programmatically without template

[flapili]

Hi,
I'm creating a framework based on FastAPI and Gino and I need to create a cli command (I'm using click) which upgrade on the fly the database.

I want avoid the use of the classic template ( env.py / ini file /ect ... )

for now I have

import pprint

from alembic.migration import MigrationContext
# from alembic.autogenerate import compare_metadata
from alembic.autogenerate import produce_migrations
# from sqlalchemy.schema import SchemaItem
# from sqlalchemy.types import TypeEngine
from sqlalchemy import create_engine, MetaData, Column, Integer, String, Table, text



engine = create_engine("sqlite://")

with engine.begin() as conn:
    conn.execute(
        text(
            """
        create table foo (
            id integer not null primary key,
            old_data varchar,
            x integer
        )"""
        )
    )

    conn.execute(
        text(
            """
        create table bar (
            data varchar
        )"""
        )
    )

metadata = MetaData()
Table(
    "foo",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("data", Integer),
    Column("x", Integer, nullable=False),
)
Table("bat", metadata, Column("info", String))

mc = MigrationContext.configure(engine.connect())

migration = produce_migrations(mc, metadata)
pprint.pprint(migration, indent=2, width=20)
pprint.pprint(migration.__dir__())
print(type(migration))

But I don't know what to do with alembic.operations.ops.MigrationScript

I'm not sure to be in the right way :(
Have a nice day!

edit:

I trying a different way, I'm inspiring by https://github.com/patiprecios/fastapi-migrations and https://github.com/miguelgrinberg/Flask-Migrate

PS C:\Users\DEVEAUX\Desktop\prog\ModularAPI> python .\modularapi\cli.py
Generating C:\Users\DEVEAUX\Desktop\prog\ModularAPI\versions\c0188f900bb9_test.py ...  FAILED
Traceback (most recent call last):
  File ".\modularapi\cli.py", line 32, in <module>
    r = command.revision(config=alembic_cfg, message="test")
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\command.py", line 221, in revision
    scripts = [script for script in revision_context.generate_scripts()]
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\command.py", line 221, in <listcomp>
    scripts = [script for script in revision_context.generate_scripts()]
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\autogenerate\api.py", line 513, in generate_scripts
    yield self._to_script(generated_revision)
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\autogenerate\api.py", line 429, in _to_script
    return self.script_directory.generate_revision(
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\script\base.py", line 640, in generate_revision
    self._generate_template(
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\script\base.py", line 496, in _generate_template
    util.status(
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\util\messaging.py", line 52, in status
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\alembic\util\pyfiles.py", line 17, in template_to_file
    template = Template(filename=template_file)
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\mako\template.py", line 348, in __init__
    module = self._compile_from_file(path, filename)
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\mako\template.py", line 429, in _compile_from_file
    data = util.read_file(filename)
  File "C:\Users\DEVEAUX\AppData\Local\Programs\Python\Python38\lib\site-packages\mako\util.py", line 383, in read_file
    fp = open(path, mode)
FileNotFoundError: [Errno 2] No such file or directory: '.\\script.py.mako

the tree:

ModularAPI
│   .env
│   poetry.lock
│   pyproject.toml
│   README.rst
│   temp
│
├───.vscode
│       settings.json
│
├───dist
│       ModularAPI-0.0.1-py3-none-any.whl
│       ModularAPI-0.0.1.tar.gz
│
├───modularapi
│   │   app.py
│   │   asgi.py
│   │   cli.py
│   │   db.py
│   │   settings.py
│   │   utils.py
│   │   venv_requirements.txt
│   │   __init__.py
│   │   
│   ├───templates
│   │   └───default
│   │           env.py
│   │           README
│   │           script.py.mako
│   │
│   └───__pycache__
│           utils.cpython-38.pyc
│
├───tests
│       test_modularapi.py
│       __init__.py
│
└───versions

and the script:

import sys
import os
import venv
import shutil
import subprocess
import time
import json
from pathlib import Path

import click
import git

from alembic import command
from alembic.config import Config


from modularapi.settings import get_setting
from modularapi.utils import _on_rmtree_error


class AlembicConfig(Config):
    def get_template_directory(self):
        pkg_dir = Path(__file__).resolve().parent
        return pkg_dir / "templates"


(Path() / "versions").mkdir(exist_ok=True)

alembic_cfg = AlembicConfig()
alembic_cfg.set_main_option("sqlalchemy.url", get_setting().PG_DNS)
alembic_cfg.set_main_option("script_location", str(Path()))
r = command.revision(config=alembic_cfg, message="test")
print(r)
exit(0)

I don't understand what is the issue :/

[zzzeek]

try looking at the example here: https://alembic.sqlalchemy.org/en/latest/cookbook.html#run-alembic-operation-objects-directly-as-in-from-autogenerate

but alembic is not really intended to automatically upgrade a database without manual intervention. you'd want to look at Django's migrations for that kind of thing.

[flapili]

thanks for your reply, after reflexion I think I'll rewrap the alembic.command package in my cli with the second way.

have you any hint about the fileNotFoundIssue ? does the error mean the versions folder must be in the templates/default dir ? (where script.py.mako is located) ?

[zzzeek]

the alembic "directory" that is configured in the alembic.ini has a mako file called script.py.mako inside of it. it looks like that directory is not configured properly.

[flapili]

I don't want to use ini file at all.

what should be the script_location key of the alembic config in the case I want to use env.py and script.py.mako in the path X but have versions folder in the path Y ?

The final goal is to have the env.py/script.py.mako in my package (in site-packages) and the versions dir at the cwd.

edit

I think I understood what is the function of templates folder,
So if I understand correclty I must init an alembic folder with command.init when I initialize my Modular project ?

[flapili]

According to your comment I continue here:

I think I missexplained my project:

the goal of my framework is allow to add modules (from github as example) to a directory, then it will load the modules

app.py:

def get_app():
    app = FastAPI()

    for path in Path("").glob("modules/*"):
        if path.is_dir():

            module_name = ".".join(path.parts[:-1] + (path.stem, "main"))
            logger.info(f"loading module {module_name}")

            try:
                module = import_module(module_name)

                try:
                    getattr(module, "load_module")(app)
                except AttributeError:
                    logger.error(f"Could not load module {module_name} ! missing entrypoint !")

            # there is no main.py in the module
            except ModuleNotFoundError:
                logger.error(f"Could not load module {module_name} missing main.py!")

    db.init_app(app)
    return app

the structure of a modular app should look like:

project_foo
│   requirements.txt
├───venv
├───venv
├───db_migrations
│   │   env.py
│   │   README
│   │   script.py.mako
│   │   __init__.py
│   │
│   └───versions
│           __init__.py
└───modules
    ├───module_a
    │       bd.py
    │       main.py
    │
    └───module_b
            bd.py
            main.py

and I have a cli which init the project:

# init <project_name>
@cli.command(name="init")
@click.argument("project_path", type=click.Path())
def cli_projet_init(project_path):
    """
    Init a new Modular project
    """

    p = Path(project_path)

    # check if the path already exists
    if p.is_dir():
        click.secho(f"{error_style} the path `{p}` already exist !")
        exit(1)

    # Initialization
    click.echo(f"{info_style} Initializing a new projet at `{p}` ...")
    (p / "modules").mkdir(parents=True)

    try:
        with open(os.devnull, "w") as f:
            with redirect_stdout(f):
                command.init(
                    config=alembic_cfg,
                    directory="db_migrations",
                    template="default",
                    package=True,
                )
    except UnboundLocalError:  # temporary du to a bug in alembic (issue 808)
        pass

    # creating the venv
    click.echo(f"{info_style} Creating the venv ...")
    venv.EnvBuilder(with_pip=True).create(p / "venv")

    # install dependancies
    click.echo(f"{info_style} Installing dependancies in the venv ...")

    # on Windows
    if os.name == "nt":
        python_path = p / "venv" / "Scripts" / "python.exe"

    # on Unix
    else:
        python_path = p / "venv" / "bin" / "python"

    subprocess.run([python_path, "-m", "pip", "install", "-U", "pip"], check=True)
    subprocess.run(
        [
            python_path,
            "-m",
            "pip",
            "install",
            "-r",
            Path(__file__).parent / "venv_requirements.txt",
        ],
        check=True,
    )

    shutil.copyfile(
        src=Path(__file__).parent / "venv_requirements.txt",
        dst=p / "requirements.txt",
    )

as you can see I don't use ini file, should I ?

[zzzeek]

the structure of a modular app should look like:

project_foo
│   requirements.txt
├───venv
├───venv
├───db_migrations
│   │   env.py
│   │   README
│   │   script.py.mako
│   │   __init__.py
│   │
│   └───versions
│           __init__.py
└───modules
    ├───module_a
    │       bd.py
    │       main.py
    │
    └───module_b
            bd.py
            main.py

when you have the alembic files as part of the modules you also have the option to load them as a package resource.

as you can see I don't use ini file, should I ?

technically no, since your code refers to a variable named "alembic_cfg" that I'm not seeing declared anywhere. however, if you are building Config programmatically that is perfectly fine, don't use the ini file. I don't think you really need the "alembic init" command either as the main job this command does is create the .ini file from programatic defaults. The rest of the flles it just copies, which you could just do with a few lines. That said, I think you should have an alembic.ini file anyway since it is a useful place for people to change config settings if they want to. Here is an example of a very widely used embedded Alembic CLI in Openstack Neutron: https://github.com/openstack/neutron/tree/master/neutron/db/migration there's an alembic.ini file that is basically empty.

[flapili]

thanks for your help, I finnaly finish the cli of my framework by using alembic.command ;)

now I want to show a warning and exit when the database is not up to date with the schema declaration,
does this recipe is suitable for this ?

[zzzeek]

yes, there's also a pypi project https://pypi.org/project/alembic-autogen-check/ that uses a slightly different pattern you can see here: https://github.com/4Catalyzer/alembic-autogen-check/blob/master/alembic_autogen_check.py they actually run the alembic autogenerate and capture the diffs using process_revision_directives. not sure if there's some advantage to doing it that way but you should be able to use compare_metadata() directly that's what it's for.

[ItsDrike]

This is what I'm doing to programatically run the migrations from the alembic-migrations directory, where it still automatically picks up any of the version files. That then allows me to use alembic with env.py and alembic.ini normally when I want to use the CLI, and use this to automatically apply the migrations when I start my project.

Any feedback is welcome.

def apply_db_migrations(db_conn: Connection) -> None:
    """Apply alembic database migrations.

    This method will first check if the database is empty (no applied alembic revisions),
    in which case, it use SQLAlchemy to create all tables and then stamp the database for alembic.

    If the database is not empty, it will apply all necessary migrations, bringing the database
    up to date with the latest revision.
    """
    # Create a standalone minimal config, that doesn't use alembic.ini
    # (we don't want to load env.py, since they do a lot of things we don't want
    # like setting up logging in a different way, ...)
    alembic_cfg = alembic.config.Config()
    alembic_cfg.set_main_option("script_location", "alembic-migrations")
    alembic_cfg.set_main_option("sqlalchemy.url", SQLALCHEMY_URL)

    script = ScriptDirectory.from_config(alembic_cfg)

    def retrieve_migrations(rev: str, context: MigrationContext) -> list[RevisionStep]:
        """Retrieve all remaining migrations to be applied to get to "head".

        The returned migrations will be the migrations that will get applied when upgrading.
        """
        migrations = script._upgrade_revs("head", rev)  # pyright: ignore[reportPrivateUsage]

        if len(migrations) > 0:
            log.info(f"Applying {len(migrations)} database migrations")
        else:
            log.debug("No database migrations to apply, database is up to date")

        return migrations

    env_context = EnvironmentContext(alembic_cfg, script)
    env_context.configure(connection=db_conn, target_metadata=Base.metadata, fn=retrieve_migrations)
    context = env_context.get_context()

    current_rev = context.get_current_revision()

    # If there is no current revision, this is a brand new database
    # instead of going through the migrations, we can instead use metadata.create_all
    # to create all tables and then stamp the database with the head revision.
    if current_rev is None:
        log.info("Performing initial database setup (creating tables)")
        Base.metadata.create_all(db_conn)
        context.stamp(script, "head")
        return

    log.debug("Checking for database migrations")
    with Operations.context(context) as _op, context.begin_transaction():
        context.run_migrations()