Forem: Shreshthajit das

Database handling using Python Alembic and SQL Alchemy

Shreshthajit das — Wed, 19 Nov 2025 19:13:09 +0000

Suppose we are working on a Python backend application, and we often make changes that require database updates. Alembic and SQLAlchemy are excellent solutions for using migrations in Python. They both provide a modern approach to creating and tracking all changes to your databases. Alembic is a database migrations tool that works together with SQLAlchemy.

We can initialize Alembic in our project through this command:

alembic init migrations

We can see a new folder called migrations under the project folder. This folder will be used by Alembic to manage the migrations, revisions, and configurations.

We need to first understand the Python SQLAlchemy and what Alembic does. What does an alembic upgrade and downgrade do?

SQLAlchemy is like an "Object Relational Mapper." When a Python class is created and defined, SQLAlchemy takes my Python class and "maps" it to a database.

Alembic works together with SQLAlchemy to modify our database without overwriting or deleting any data. Suppose we modified our schema and want to reflect the new changes. Now, Alembic will look at the old database, then compare it to the new one we generate, and will perform the changes on our target database table. Here's an alembic migration demo file:

"""Add new column 'status' to users table

Revision ID: 123456789abc
Revises: abcdef123456
Create Date: 2025-12-02 18:00:00
"""

from alembic import op
import sqlalchemy as sa

# revision identifiers, used by Alembic.
revision = "123456789abc"
down_revision = "abcdef123456"
branch_labels = None
depends_on = None


def upgrade() -> None:
    """
    Upgrade migrations: apply changes to the database schema.
    Example: Adding a new column.
    """
    op.add_column(
        "users",
        sa.Column("status", sa.String(length=20), nullable=True),
    )


def downgrade() -> None:
    """
    Downgrade migrations: reverse the upgrade changes.
    Example: Removing the column added in upgrade().
    """
    op.drop_column("users", "status")

Let's understand the alembic upgrade and downgrade command:

Upgrade the database forward by exactly one migration step from the current version. Suppose our migrations in version files are the following:
A -> B -> C

Our current head in versions files is C but B in the database. Then type this command:

alembic upgrade +1

will move us from B → C.

Suppose our current version is C and we want to downgrade the database back by one migration step. Then we can type the following command:

alembic downgrade -1

moves us from C → B

Let's say I am working in a branch where I have made some changes, and these changes affected some tables in the database by inserting a new column. Now I want to work on another branch where this feature is not implemented. In that case, if I run the project, I will get an error from the backend since the database migration occurred when I worked on the previous feature. And if we downgrade, then it will work.

Every time we make changes to the schema, Alembic will autogenerate a file with a version number that stores all the changes made to the database, as well as a downgrade command to revert changes.

Sometimes, more than one alembic file is generated, which is sometimes not required at all. We can simply merge them into one. To do this, we can follow the process described below.

First, we can type this command:

alembic history

We will see something like:

a1->b2 -> c3 -> d4 -> e5 -> f6 -> g7 -> h8 -> i9 -> j10 (head)
a1 (base).

Suppose we want to merge h8,i9 and j10 into one migration file.

We will type the following command:

alembic downgrade g7

g7 will be the head now.

Now we will move these three files into another backup folder. And then run this command:

alembic revision --autogenerate -m "squashed_h8_i9_j10_migrations"

Now we can see a new migration file generated.
Then we will give this command to upgrade the head:

alembic upgrade head

There is another approach that we can also follow. If we open any migration file, we will see that every file has a downgrade and an upgrade function. So, we can generate a new file first:

alembic revision --autogenerate -m "combined-features"

Then we can simply take the code of the upgrade and downgrade functions from all the files that we want to remove and update this new file with this code. Then change the downrevision of that file to the revision of the previous file, which is the HEAD, and then do the upgrade.

Sometimes it fails because Alembic still references a missing revision in the database. We can force stamp with --revision that exists:

alembic stamp 26429623e713

Let's discuss some issues.

Scenario 1:
Sometimes Alembic throws an error like the head (b886dffa554d) does not exist, or the requested revision was not found. This usually happens when the database has an old or incorrect revision stored in alembic_version.

But our codebase (the alembic/versions directory) now uses a different head, such as 26429623e713.

To fix the mismatch, we can manually update the revision stored in our database so it matches the latest head from our codebase.

We can safely correct this using the following SQL:

UPDATE alembic_version
SET version_num = '26429623e713'
WHERE version_num = 'b886dffa554d';

To check the version

SELECT * FROM alembic_version;

Scenario 2:
Suppose there are 2 developers, dev1 and dev2. They started working on two separate branches with 2 new features, corresponding to cloning the main branch. Suppose both made changes to their branch.

In main branch: a->b(head)

dev1: a->b->d(head)
dev2: a->b->c(head)

While merging :

dev1: a->b->d(head) merged to main

dev2: a->b->c(head) [error: version d doesn’t exist]

So, first we can rebase the dev1 branch now that the head is d. While adding dev2's feature, we can make the dev2's head, which is c, next to dev 1, which is d now. And push the changes to the dev1 branch. And then merge the dev1 PR. But this is sometimes problematic. Because we are rewriting migration history, which is sometimes dangerous.

If migrations modify the same tables/columns, rebase may cause:

inconsistent database states
duplicated constraints
partial schema drift
Alembic does not support manually reordering migrations to resolve divergence.

So, here is another solution:

If dev1 merges first, the main branch becomes:

a -> b -> d (head)

Next pull the latest main into dev2’s branch:

git checkout dev2
git pull origin main

Now dev2 has both migrations:

a -> b -> d (main)
a -> b -> c (dev2)

Alembic will detect two heads: d, c.

Run

alembic heads

Create a merge migration:

alembic merge -m "Merge dev1 and dev2 migrations" d c

A new merge file is created. Now we can push it to dev2's branch and create a PR.

Understanding Asyncio and Multithreading in Python

Shreshthajit das — Sun, 26 Oct 2025 05:42:02 +0000

When we think about synchronous program, it tells us that any task must finish before the next one starts. An asynchronous program can toggle between tasks so that tasks can run concurrently without blocking each other.

In Python, let's first try to understand the concepts of a coroutine and a subroutine. We are all familiar with a function, which is also known as a subroutine, procedure, sub-process, etc. A subroutine is just a normal Python function defined with def.

On the other hand, a coroutine is something that is defined with async def and can be paused with await or yield. A normal function in Python becomes a coroutine when it is defined with the async def syntax.

The async def lets the current function temporarily pause its execution while the execution of respective (I/O, network requests, etc.) operations is in progress.

It is important to note that coroutines do not make the code multi-threaded; rather, coroutines run in an event loop that executes in a single thread. Upon the usage of async def, the said function yields a coroutine object. When the await keyword is encountered, the current coroutine is paused, and control is passed back to the event loop.

The CPU is less utilized (or might be free) when I/O (or similar operations) are in progress. For instance, copying data to an external hard drive is an I/O operation where the CPU only initiates and accepts the I/O requests. The CPU can be better utilized in such cases for performing other tasks! The event loop continuously monitors the awaitable (coroutine, Task, or Future) until it is completed. Once the execution of the awaitable or the newly picked-up task is complete, the event loop restores the execution of the paused coroutine.

See the code below:

import asyncio
import sys
import time
from datetime import datetime

async def task1():
   print("Enter " + sys._getframe().f_code.co_name + " " + str(datetime.now().time()))    # Get function name
   await asyncio.sleep(2)    # Could be an I/O operation, network request, database operation, and more
   ret_info = await task2()
   print("After sleep " + sys._getframe().f_code.co_name + " " + str(datetime.now().time()))
   return "task1"

async def task2():
   print("Enter " + sys._getframe().f_code.co_name + " " + str(datetime.now().time()))
   await asyncio.sleep(2)
   print("After sleep " + sys._getframe().f_code.co_name + " " + str(datetime.now().time()))
   return "task2"

async def main():
   print("Enter main")
   start_time = time.perf_counter()
   ret_info = await task1()
   print(f"Data received from the task1: {ret_info}" + " " + str(datetime.now().time()))
   ret_info = await task2()
   print(f"Data received from the task2: {ret_info}" + " " + str(datetime.now().time()))
   end_time = time.perf_counter()
   print("Exit main")
   print(f'It took {round(end_time - start_time,0)} second(s) to complete.')


if __name__ == '__main__':
   # main coroutine
   asyncio.run(main())

Here is the output:

Enter main
Enter task1 10:27:58.983919
Enter task2 10:28:00.985972
After sleep task2 10:28:02.988581
After sleep task1 10:28:02.988645
Data received from the task1: task1 10:28:02.988658
Enter task2 10:28:02.988674
After sleep task2 10:28:04.991250
Data received from the task2: task2 10:28:04.991315
Exit main
It took 6.0 seconds to complete.

Here's the explanation:

asyncio.run() starts an event loop and runs the coroutine main().So execution enters the main() coroutine
Prints "Enter main".start_time records the precise start time for measuring total duration.
Call and await task1():
- Control passes to task1().
- Because of await, main() is paused until task1() completes.
Inside task1(): Prints "Enter task1" along with the current time. asyncio.sleep(2) pauses task1 for 2 seconds asynchronously.
During this time, other async tasks could run — but since main() is just waiting, nothing else does.
After 2 seconds, task1 resumes: Calls task2() and waits for it to complete.
Inside task2() (called from task1):
- Prints entry message.
- Sleeps for another 2 seconds.
- Prints after-sleep message and returns "task2".
Back to task1: Once task2() finishes, task1() prints and returns "task1".
Back to main(): Prints: "Data received from the task1: task1". Then it runs another: Calls and awaits task2() again (same as before: 2-second delay).
After task2() returns:
- Prints the result of task2.
- Ends program and prints total duration.

The task1() and task2() are defined as asynchronous functions (or coroutines). During the execution of task1(), an await keyword is encountered with an async sleep of 2 seconds. This pauses the coroutine and yields control back to the event loop.

With this, the execution of task1() is paused until the completion of task2() coroutine. Post the execution of task2(), the execution of task1() coroutine is resumed, and the return value of task1 is printed on the terminal.

To sum up, asyncio is built on coroutines managed by the event loop, with tasks scheduling them and futures holding results.

Multithreading

Python achieves concurrency via context switching, which involves frequently switching between threads on single-core CPUs, giving the illusion of parallel execution or multitasking.

Multiple threads facilitate the execution of background tasks without blocking the main program.

Consider the diagram below to understand how multiple threads exist in memory:

Before moving ahead, it’s important to note that Python threading is different from threading in other languages like Java, C#, or Go, where the threads can execute simultaneously on a multicore processor. Due to the design limitations of the Python interpreter(CPython), only one thread can run at a time.

Actually, threads are managed by the Global Interpreter Lock (GIL), which allows only one thread to execute Python bytecode at a time, even on multi-core systems.

What’s really interesting is, threads in Python can coexist (run concurrently), and the interpreter switches between them rapidly (context switching).Example: Thread A runs a bit, then Thread B runs a bit, then back to A. But they cannot run at the exact same time on multiple cores (because the GIL allows only one to execute Python bytecode at once). So, threads are concurrent, not parallel, in CPU-bound tasks.

To fully use multiple CPU cores in Python, use the multiprocessing module, which creates separate processes, each with its own GIL. Now each process runs independently, true parallelism.

Here, we can create three threads that will independently download the three webpages concurrently and take varying amounts of time depending upon network conditions.
The main thread will wait until all three threads have finished downloading since the join method of all the threads is set in the main thread.
Once all three downloads are finished, the data is given to the create_single_webpage() function to create a single webpage.
If the join() method is not used, the main thread will immediately run the create_single_webpage() function before the downloads are finished and resulting in an error.

Advantages of Threading

Threaded programs run faster on computer systems that have multiple CPUs.
Threads of a process can share the memory of global variables. If the variable's value is changed in one thread, it is applicable to all threads.

In the next part of this series, I’ll dive deeper into threads and how they differ from asyncio tasks.