Alembic Operations

Jan 24th, 2020 - written by Kimserey with .

A few weeks ago we looked into Alembic to perform database migrations. We looked into how to install it and the commands it provides to manage databases. In today’s post we will look at the main DDL operations we would use to manage databases.

Create Table
Modify Column
Modify Constraint
Modify Index

Create Table

Following up our last post, we’ve already seen a create table migration where SQLAlchemy (sa) is used to define the table schema.

def upgrade():
    op.create_table(
        'user',
        sa.Column('id', sa.Integer, primary_key=True),
        sa.Column('username', sa.String(), nullable=False),
        sa.Column('email', sa.String(), nullable=False)
    )

This will create a user table on the database at the path specified on the alembic config.

Modify Column

Adding a column can be done using the add_column operation:

add_column(table_name, column, schema=None)

def upgrade():
    op.add_column('user', sa.Column('age', sa.Integer()))

In constract dropping a column is not supported at an operation level but rather as a batch operation level. Batch mode is a special operation from Alembic allowing a series of operatin to take place as a batch, executing the different manoeuvers necessary for recreating tables due to SQLite restrictions where data are copied in temp table and the table is recreated.

with op.batch_alter_table("user") as batch_op:
    batch_op.drop_column('age')

Similarly updating a column or renaming a column is handled in a batch operation.

alter_column(column_name, nullable=None, comment=False, server_default=False, new_column_name=None, type_=None, existing_type=None, existing_server_default=False, existing_nullable=None, existing_comment=None, **kw)

def upgrade():
    with op.batch_alter_table("user") as batch_op:
        batch_op.alter_column("username", new_column_name="name")

Modify Constraint

The next operation that we would often use is updating constraints. Constraints can come in form of Primary or Foreign key constraints or Check constraints.

Primary and Foreign keys can be added at table creation from sa.Column schema:

def upgrade():
    op.create_table(
        'user_profile',
        sa.Column('id', sa.Integer, sa.ForeignKey('user.id'), primary_key=True),
        sa.Column('username', sa.String(), nullable=False),
        sa.Column('email', sa.String(), nullable=False)
    )

This created the following table with two constraints:

CREATE TABLE user_profile (
    id INTEGER NOT NULL,
    username VARCHAR NOT NULL,
    email VARCHAR NOT NULL,
    PRIMARY KEY (id),
    FOREIGN KEY(id) REFERENCES user (id)
)

One important aspect of adding the constraints from the schema is that the PRIMARY KEY and FOREIGN KEY constraints are unnamed. If we want to delete the constraints, we will need to use the naming_convention:

def upgrade():
    naming_convention={
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s"
    }
    
    with op.batch_alter_table("user_profile", naming_convention=naming_convention) as batch_op:
        batch_op.drop_constraint('pk_user_profile')
        batch_op.drop_constraint('fk_user_profile_id_user')

If we were to create constraints out of the table creation, we would be able to name the constraint directly hence would be able to do without naming_convention using create_primary_key and create_foreign_key:

create_primary_key(constraint_name, columns)

create_foreign_key(constraint_name, referent_table, local_cols, remote_cols, referent_schema=None, onupdate=None, ondelete=None, deferrable=None, initially=None, match=None, **dialect_kw)

def upgrade():
    with op.batch_alter_table("user_profile") as batch_op:
        batch_op.create_primary_key("pk_user_profile", ["id"])
        batch_op.create_foreign_key("fk_user_profile_id_user", "user",["id"], ["id"])

The difference here is that we have added the constraints separatly and named them, if we were to check the schema we would see the following:

CREATE TABLE "user_profile" (
    id INTEGER NOT NULL,
    username VARCHAR NOT NULL,
    email VARCHAR NOT NULL,
    CONSTRAINT pk_user_profile PRIMARY KEY (id),
    CONSTRAINT fk_user_profile_id_user FOREIGN KEY(id) REFERENCES user (id)
)

In order to create check constraints we use create_check_constraint:

create_check_constraint(constraint_name, condition, **kw)

def upgrade():
    with op.batch_alter_table("user_profile") as batch_op:
        batch_op.create_check_constraint(
            "ck_username_len",
            func.length(column('username')) > 5
        )
        batch_op.create_check_constraint(
            "ck_email_len",
            func.length(column('email')) > 5
        )

func executes an underlying SQLite function, here length on column username and email generating the following constraint:

CONSTRAINT ck_username_len CHECK (length(username) > 5)

One of the issues with check contraints is that the table reflection for SQLite does not include the check constraints therefore if we wanted to do any batch operations which would result in a temp table created, we would need to provide the table definition explicitely.

def updgrade():
    with op.batch_alter_table("user_profile") as batch_op:
        batch_op.drop_constraint("ck_username_len")
        batch_op.drop_constraint("ck_email_len")

would yield the following error:

raise ValueError("No such constraint: '%s'" % const.name)
ValueError: No such constraint: 'ck_username_len'

By providing the table definition through copy_from we are then able to delete the ck_username_len constraint:

def upgrade():
    table = sa.Table(
        'user_profile',
        sa.MetaData(),
        sa.Column('id', sa.Integer, sa.ForeignKey('user.id'), primary_key=True),
        sa.Column('username', sa.String(), nullable=False),
        sa.Column('email', sa.String(), nullable=False),
        sa.CheckConstraint("", name="ck_username_len"),
        sa.CheckConstraint(func.length(column('email')) > 5, name="ck_email_len"),
    )

    with op.batch_alter_table("user_profile", copy_from=table) as batch_op:
        batch_op.drop_constraint("ck_username_len")

One important point to note is that copy_from defines the new table schema. Therefore we have to provide back the exact constraint for other constraints untouched, for example ck_email_len is provided again. Also because it is a new table, the primary key and foreign key are back to being unnamed constraints, if we wanted to maintain the exact same schema, we would need to add the operation to name them as part of this migration.

Here’s the resulting schema:

CREATE TABLE "user_profile" (
    id INTEGER NOT NULL,
    username VARCHAR NOT NULL,
    email VARCHAR NOT NULL,
    PRIMARY KEY (id),
    CONSTRAINT ck_email_len CHECK (length(email) > 5),
    FOREIGN KEY(id) REFERENCES user (id)
)

Modify Index

Lastly adding a new index can be done with the operation:

create_index(index_name, table_name, columns, schema=None, unique=False, **kw)

We can use it as followed, where columns is an array of column name:

def upgrade():
    op.create_index("idx_username", "user_profile", ["username"], unique=True)

This produces the following schema:

CREATE UNIQUE INDEX idx_username ON user_profile (username)

And similarly dropping an index can be done with drop_index:

def downgrade():
    op.drop_index("idx_username", "user_profile")

And that concludes today’s post!

Conclusion

Today we looked in more details in Alembic operations, creating migrations for different scenarios. We started by looking at how to create table. Then moved on to how to modify columns and started to explore the different approach to handle that for SQLite. We then moved on to see how to modify constraints, and looked at some important points to consider when doing so like specifying the temp table schema. Lastly we looked into creating indexes. I hope you liked this post and I see you on the next one!