SQLAlchemy sync/async example

This repository shows a minimal SQLAlchemy 2.0 setup that supports both synchronous and asynchronous data access. It ships with SQLite defaults, but the connection string in env.py can be swapped for Postgres or MySQL. The code is small on purpose so you can copy the bits you need.

Contents

• main.py: runs the async demo end-to-end.
• application.py: orchestrates async CRUD calls.
• composition_root.py: wires the async context and repository.
• database/context.py: sync engine and session factory.
• database/context_async.py: async engine and session factory.
• database/models.py: SQLAlchemy models for users, tax accounts, and salaries.
• repositories/user_repository.py: sync CRUD helpers.
• repositories/user_repository_async.py: async CRUD helpers.
• database/migrations: Alembic scaffold.

Requirements

• Python 3.13+
• SQLite (bundled) or another DB driver for your chosen backend
• uv (fast installer/runner): https://docs.astral.sh/uv/

Setup (uv)

1. Create and activate a virtual environment:

uv venv
source .venv/bin/activate

2. Install dependencies from pyproject:

uv sync

3. Adjust the connection string in env.py if you want Postgres or MySQL. Example:

DB_CONNECTION_STRING = "postgresql+asyncpg://user:password@localhost:5432/dbname"

Run the async demo

python main.py

The script will create a user, read it, update it, delete it, and dispose the async engine. All operations are performed through the async repository and sessionmaker.

Using the async stack

from database.context_async import DbContextAsync
from repositories.user_repository_async import UserRepositoryAsync
from database.models import UserDb

db = DbContextAsync("sqlite+aiosqlite:///example.db")
repo = UserRepositoryAsync(db)

async def create_user():
	user = UserDb(name="Jane", email="jane@example.com")
	await repo.create(user)
	return await repo.get_by_id(user.id)

Using the sync stack

from database.context import DbContext
from repositories.user_repository import UserRepository
from database.models import UserDb

db = DbContext("sqlite:///example.db")
repo = UserRepository(db)

user = repo.create(UserDb(name="John", email="john@example.com"))
print(repo.get_by_id(user.id))

Running migrations (optional) Alembic is scaffolded in database/migrations. To generate a revision after editing models, run:

alembic revision --autogenerate -m "describe change"
alembic upgrade head

Alembic runs with the synchronous SQLAlchemy engine. If your app uses an async URL (for example sqlite+aiosqlite:// or postgresql+asyncpg://), configure a separate sync URL for migrations (for example sqlite:/// or postgresql+psycopg://). Update database/migrations/env.py to point DB_CONNECTION_STRING at the sync driver so migrations can connect.

License MIT. See LICENSE when publishing to GitHub.