Skip to content

Alembic wrapper#

Alembic is great but it has a configuration problem. While you web application config is probably one or more python files and parameters loaded from environment variables and other sources, Alembic needs a static alembic.ini file. You are suppose to edit the almost-undocumented env.py file to customize it to your application.

The Alembic wrapper class aims to simplify that set up so you can just use your application config, without any more config files to maintain. The actual database migrations are still handled by Alembic so you get exactly the same functionality.

Set up#

The Alembic() class require two arguments: A SQLAlchemy() instance and the path of the folder that will contain the migrations.

from sqla_wrapper import Alembic, SQLAlchemy

db = SQLAlchemy()
alembic = Alembic(db, "db/migrations")

If the migrations folder doesn’t exists, it will be created.

CLI integrations#

The only downside is that you can’t use the alembic command-line tool anymore. Instead, all the usual Alembic command are be available as methods of the wrapper instance and you need to integrate them with your framework/application CLI.

Is easier than it sounds, specially because the wrapper comes with one-line methods to extend Click (the CLI used by Flask by default) and pyCEO (arguably, the best CLI ever made).

Integrating with Flask Click#

from flask import Flask

db = SQLAlchemy()
alembic = Alembic()

app = Flask(__name__)
app.cli.add_command(alembic.get_flask_cli("db"))

Integrating with Click#

import click

db = SQLAlchemy()
alembic = Alembic()

@click.group()
def cli():
    pass

cli.add_command(alembic.get_click_cli("db"))

Integrating with pyCEO#

from pyceo import Cli

db = SQLAlchemy()
alembic = Alembic()

class Manage(Cli):
  db = alembic.get_pyceo_cli("db")

cli = Manage()

API#

For a more in-depth understanding of these methods and the extra options, you can read the documentation for the Alembic config.

class sqla_wrapper.Alembic(db, path=’db/migrations’, **options)
revision(self, message, *, empty=False, parent=’head’)

Create a new revision. Auto-generate operations by comparing models and database.

Arguments:

  • message: Revision message.
  • empty: Generate just an empty migration file, not the operations.
  • parent: Parent revision of this new revision.
upgrade(self, target=’head’, *, sql=False, **kwargs)

Run migrations to upgrade database.

Arguments:

  • target: Revision target or “from:to” range if sql=True. “head” by default.
  • sql: Don’t emit SQL to database, dump to standard output instead.
  • **kwargs: Optional arguments. If these are passed, they are sent directly to the upgrade() functions within each revision file. To use, modify the script.py.makotemplate file so that the upgrade() functions can accept arguments.
downgrade(self, target=’-1’, *, sql=False, **kwargs)

Run migrations to downgrade database.

Arguments:

  • target: Revision target as an integer relative to the current state (e.g.: “-1”), or as a “from:to” range if sql=True. “-1” by default.
  • sql: Don’t emit SQL to database, dump to standard output instead.
  • **kwargs: Optional arguments. If these are passed, they are sent directly to the downgrade() functions within each revision file. To use, modify the script.py.mako template file so that the downgrade() functions can accept arguments.
get_history(self, *, start=None, end=None)

Get the list of revisions in chronological order. You can optionally specify the range of revisions to return.

Arguments:

  • start: From this revision (including it.)
  • end: To this revision (including it.)
history(self, *, verbose=False, start=’base’, end=’heads’)

Print the list of revisions in chronological order. You can optionally specify the range of revisions to return.

Arguments:

  • verbose: If True, shows also the path and the docstring of each revision file.
  • start: Optional starting revision (including it.)
  • end: Optional end revision (including it.)
stamp(self, target=’head’, *, sql=False, purge=False)

Set the given revision in the revision table. Don’t run migrations.

Arguments:

  • target: The target revision; “head” by default.
  • sql: Don’t emit SQL to the database, dump to the standard output instead.
  • purge: Delete all entries in the version table before stamping.
get_current(self)

Get the last revision applied.

current(self, verbose=False)

Print the latest revision(s) applied.

Arguments:

  • verbose: If True, shows also the path and the docstring of the revision file.
get_head(self)

Get the latest revision.

head(self, verbose=False)

Print the latest revision.

Arguments:

  • verbose: If True, shows also the path and the docstring of the revision file.
init(self, path)

Creates a new migration folder with a script.py.mako template file. It doesn’t fail if the folder or file already exists.

Arguments:

  • path: Target folder.
create_all(self)

Create all the tables from the current models and stamp the latest revision without running any migration.

rev_id(self)

Generate a unique id for a revision.

By default this uses alembic.util.rev_id. Override this method to change it.

get_pyceo_cli(self)
get_click_cli(self, name=’db’)
get_flask_cli(self, name=’db’)
Back to top