sqlalchemy-dlock

A distributed lock library based on databases and SQLAlchemy.

sqlalchemy-dlock provides distributed locking capabilities using your existing database infrastructure—no additional services like Redis or ZooKeeper required. It currently supports:

psycopg refers to Psycopg 3. CI tests it in both synchronous and asynchronous modes. The integration matrix covers MySQL, PostgreSQL, and Microsoft SQL Server on Python 3.10 through 3.14.

⚠️ Oracle Not Tested: Oracle Database Free (23c/23ai) does NOT support DBMS_LOCK.REQUEST. We do NOT test Oracle in CI or integration tests. Use with Oracle Enterprise/Standard Edition at your own risk.

Why sqlalchemy-dlock?

Distributed locks coordinate access to shared resources across multiple processes or servers. Here's how database-based locking compares to other approaches:

sqlalchemy-dlock is ideal for:

• Projects already using MySQL, MariaDB, MSSQL, Oracle, or PostgreSQL
• Teams wanting zero additional infrastructure
• Low to medium concurrency distributed synchronization
• Applications requiring strong consistency guarantees

Not recommended for:

• High-concurrency scenarios (consider Redis instead)
• Situations sensitive to database load

Quick Start

Installation

pip install sqlalchemy-dlock

Requirements:

• Python 3.10+
• SQLAlchemy 1.4.3+ or 2.x
• Appropriate database driver for your database (see below)

Database Drivers

This library requires a database driver to be installed separately. Since you're already using SQLAlchemy, you likely have the appropriate driver installed. For a complete list of SQLAlchemy-supported drivers, see the SQLAlchemy Dialects documentation.

ℹ️ Notes:

• MSSQL: The pyodbc driver requires the Microsoft ODBC driver to be installed on your system. On Ubuntu/Debian:

  sudo ACCEPT_EULA=Y apt-get install -y msodbcsql18
  

• Oracle: Oracle Database Free (23c/23ai) does NOT support DBMS_LOCK.REQUEST which is required for distributed lock functionality. For production use with Oracle, a full Oracle Database (Enterprise/Standard Edition) installation is required.

Basic Usage

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")

with engine.connect() as conn:
    # Create a lock
    lock = create_sadlock(conn, "my-resource-key")

    # Acquire the lock
    lock.acquire()
    assert lock.locked

    # Release the lock
    lock.release()
    assert not lock.locked

Using Context Managers

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")

with engine.connect() as conn:
    # Automatically acquires and releases the lock
    with create_sadlock(conn, "my-resource-key") as lock:
        assert lock.locked
        # Your critical section here

    # Lock is automatically released
    assert not lock.locked

With Timeout

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")

with engine.connect() as conn:
    try:
        # Raises TimeoutError if lock cannot be acquired within 5 seconds
        with create_sadlock(conn, "my-resource-key", contextual_timeout=5) as lock:
            pass
    except TimeoutError:
        print("Could not acquire lock - resource is busy")

Key Conversion

Each database backend converts the supplied key to the type required by its locking function. The converted value is available through lock.actual_key:

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")

with engine.connect() as conn:
    lock = create_sadlock(conn, "42", convert=int)
    assert lock.actual_key == 42

Use the convert argument when the backend's default conversion does not match your key scheme. The former lock.key alias was removed in v0.9.0; use lock.actual_key instead.

Common Use Cases

Use Case 1: Preventing Duplicate Task Execution

Prevent multiple workers from processing the same task simultaneously:

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock


def process_monthly_billing(user_id: int):
    engine = create_engine("postgresql://user:pass@localhost/db")
    with engine.connect() as conn:
        # Ensure billing for a user is only processed once at a time
        lock_key = f"billing:user:{user_id}"
        with create_sadlock(conn, lock_key, contextual_timeout=0):
            # If another worker is already processing this user's billing,
            # this will fail immediately (timeout=0)
            perform_billing_calculation(user_id)
            send_bill(user_id)

Use Case 2: API Rate Limiting & Debouncing

Prevent simultaneous expensive operations on the same resource:

from fastapi import FastAPI, HTTPException
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy_dlock import create_async_sadlock

app = FastAPI()
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)


@app.post("/api/resources/{resource_id}/export")
async def export_resource(resource_id: str):
    async with AsyncSessionLocal() as session:
        # Try to acquire lock without blocking
        lock = create_async_sadlock(session, f"export:{resource_id}")
        acquired = await lock.acquire(block=False)
        if not acquired:
            raise HTTPException(status_code=409, detail="Export already in progress")

        try:
            return await perform_export(resource_id)
        finally:
            await lock.release()

Use Case 3: Scheduled Job Coordination

Ensure scheduled jobs don't overlap across multiple servers:

from apscheduler.schedulers.background import BackgroundScheduler
from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock


def data_sync_job():
    engine = create_engine("mysql://user:pass@localhost/db")
    with engine.connect() as conn:
        lock_key = "scheduled-job:data-sync"

        # Only proceed if no other server is running this job
        lock = create_sadlock(conn, lock_key, contextual_timeout=60)
        with lock:
            perform_data_sync()


scheduler = BackgroundScheduler()
scheduler.add_job(data_sync_job, "interval", minutes=30)
scheduler.start()

Use Case 4: Decorator Pattern for Clean Code

Create a reusable decorator for locking functions:

from functools import wraps
from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock


def with_db_lock(key_func, timeout=None):
    """Decorator that acquires a database lock before executing the function."""

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            engine = create_engine("postgresql://user:pass@localhost/db")
            lock_key = key_func(*args, **kwargs)

            with engine.connect() as conn:
                with create_sadlock(conn, lock_key, contextual_timeout=timeout):
                    return func(*args, **kwargs)

        return wrapper

    return decorator


# Usage
@with_db_lock(lambda user_id: f"user:update:{user_id}", timeout=10)
def update_user_profile(user_id: int, profile_data: dict):
    # This function is protected from concurrent execution
    # for the same user_id
    ...

Working with SQLAlchemy ORM

Using locks with SQLAlchemy ORM sessions:

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")
Session = sessionmaker(bind=engine)

with Session() as session:
    with create_sadlock(session, "my-resource-key") as lock:
        # Use the session within the locked context
        user = session.query(User).get(user_id)
        user.balance += 100
        session.commit()

Asynchronous I/O Support

Full async/await support for asynchronous applications:

from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy_dlock import create_async_sadlock

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")


async def main():
    async with engine.connect() as conn:
        async with create_async_sadlock(conn, "my-resource-key") as lock:
            assert lock.locked
            # Your async critical section here
        assert not lock.locked

Supported async drivers:

• MySQL: aiomysql
• PostgreSQL: asyncpg, psycopg (v3+)

PostgreSQL Lock Types

PostgreSQL provides multiple advisory lock types. Choose based on your scenario:

Example: Transaction-Level Lock

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")

with engine.connect() as conn:
    # Transaction-level lock - automatically released on commit/rollback
    with create_sadlock(conn, "my-key", xact=True) as lock:
        conn.execute(text("INSERT INTO ..."))
        conn.commit()  # Lock is released here

Example: Shared Lock for Read-Heavy Workloads

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("postgresql://user:pass@localhost/db")


# Multiple readers can hold shared locks simultaneously
def read_resource(resource_id: str):
    with engine.connect() as conn:
        with create_sadlock(conn, f"resource:{resource_id}", shared=True):
            return conn.execute(text("SELECT * FROM resources WHERE id = :id"), {"id": resource_id})


# Writers need exclusive locks
def write_resource(resource_id: str, data: dict):
    with engine.connect() as conn:
        # This will wait for all shared locks to be released
        with create_sadlock(conn, f"resource:{resource_id}") as lock:
            conn.execute(text("UPDATE resources SET ..."))
            conn.commit()

MSSQL Lock Types

SQL Server's sp_getapplock supports multiple lock modes:

Example: Exclusive Lock for Writing (Default)

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("mssql+pyodbc://user:pass@localhost/db")

with engine.connect() as conn:
    # Exclusive lock for writing (default)
    with create_sadlock(conn, "my-resource") as lock:
        conn.execute(text("UPDATE resources SET ..."))

Example: Shared Lock for Reading

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("mssql+pyodbc://user:pass@localhost/db")


# Multiple readers can hold shared locks simultaneously
def read_resource(resource_id: str):
    with engine.connect() as conn:
        with create_sadlock(conn, f"resource:{resource_id}", shared=True):
            return conn.execute(text("SELECT * FROM resources WHERE id = :id"), {"id": resource_id})

Example: Update Lock for Read-Then-Write Patterns

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("mssql+pyodbc://user:pass@localhost/db")

with engine.connect() as conn:
    # Update lock - compatible with shared locks, used for read-then-write patterns
    with create_sadlock(conn, "my-resource", update=True) as lock:
        data = conn.execute(text("SELECT * FROM resources WHERE id = :id"), {"id": resource_id})
        # Perform read operations
        # Then upgrade to exclusive lock for writing
        conn.execute(text("UPDATE resources SET ..."))

Oracle Lock Types

Oracle's DBMS_LOCK.REQUEST supports 6 lock modes with different compatibility:

Example: Exclusive Lock for Writing (Default)

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("oracle+oracledb://user:pass@localhost/db")

with engine.connect() as conn:
    # Exclusive lock for writing (default)
    with create_sadlock(conn, "my-resource") as lock:
        conn.execute(text("UPDATE resources SET ..."))

Example: Shared Lock for Reading

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("oracle+oracledb://user:pass@localhost/db")


# Multiple readers can hold shared locks simultaneously
def read_resource(resource_id: str):
    with engine.connect() as conn:
        with create_sadlock(conn, f"resource:{resource_id}", lock_mode="S"):
            return conn.execute(text("SELECT * FROM resources WHERE id = :id"), {"id": resource_id})

Example: Transaction-Level Lock

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("oracle+oracledb://user:pass@localhost/db")

with engine.connect() as conn:
    # Transaction-level lock - automatically released on commit/rollback
    with create_sadlock(conn, "my-resource", release_on_commit=True) as lock:
        conn.execute(text("INSERT INTO ..."))
        conn.commit()  # Lock is released here

Example: Using Integer Lock ID Directly

from sqlalchemy import create_engine
from sqlalchemy_dlock import create_sadlock

engine = create_engine("oracle+oracledb://user:pass@localhost/db")

with engine.connect() as conn:
    # Direct integer lock ID (no hashing needed)
    with create_sadlock(conn, 12345, lock_mode="X") as lock:
        # Direct lock ID usage
        pass

Note: String keys are converted to integer IDs using blake2b hash (similar to PostgreSQL).

Performance Considerations

• Database lock operations require network round-trips and are slower than in-memory solutions like Redis
• PostgreSQL timeout is implemented through polling and may have ~1 second variance
• Consider your concurrency requirements before choosing database-based locks

Thread Safety

• Lock objects are thread-local and cannot be safely passed between threads
• Each thread must create its own lock instance
• Cross-process/cross-server locking works normally

MySQL-Specific Behavior

⚠️ Warning: MySQL allows acquiring the same named lock multiple times on the same connection. This can lead to unexpected cascading locks:

# DANGER: On MySQL, the second acquisition succeeds immediately
with create_sadlock(conn, "my-key") as lock1:
    # This immediately returns without waiting - no real mutual exclusion!
    with create_sadlock(conn, "my-key") as lock2:
        pass

To avoid this, use separate connections or implement additional checking logic.

Lock Lifetime

• Locks are tied to your database connection
• Closing a connection releases all associated locks
• Properly manage Connection/Session lifecycle to avoid accidental lock releases

FAQ

Q: What happens if the database goes down?

A: Locks are automatically released when the database connection is lost. This is intentional behavior to prevent deadlocks.

Q: Can I pass a lock object between threads?

A: No. Lock objects are thread-local for safety reasons. Each thread should create its own lock instance pointing to the same lock key.

Q: How do I choose between MySQL and PostgreSQL?

A: Both are fully supported. PostgreSQL offers more lock types (shared/transaction-level), while MySQL's implementation is simpler.

Q: Are locks inherited by child processes?

A: No. Child processes must establish their own database connections and create new lock objects.

Q: How can I debug lock status?

A:

• MySQL: SELECT * FROM performance_schema.metadata_locks;
• PostgreSQL: SELECT * FROM pg_locks WHERE locktype = 'advisory';

Q: What's the maximum lock key size?

A:

• MySQL: 64 characters
• MSSQL: 255 characters
• PostgreSQL: Keys are converted to 64-bit integers via BLAKE2b hash
• Oracle: Keys are converted to integers (0-1073741823) via BLAKE2b hash

Q: Can I use this with SQLite?

A: No. SQLite does not support the same named/advisory lock mechanisms as MySQL or PostgreSQL.

Testing

The following database drivers are tested in CI:

MySQL:

• mysqlclient (synchronous)
• PyMySQL (synchronous and used by aiomysql)
• aiomysql (asynchronous)
• asyncmy (asynchronous)

PostgreSQL:

• psycopg2 (synchronous)
• psycopg (v3, synchronous and asynchronous)
• asyncpg (asynchronous)

Microsoft SQL Server:

• pyodbc (synchronous)
• aioodbc (asynchronous)

Oracle is not tested in CI because Oracle Database Free does not provide DBMS_LOCK.REQUEST.

Running Tests Locally

1. Install the project with development dependencies:

uv sync --group test
uv pip install mysqlclient aiomysql asyncmy "PyMySQL~=1.1.0" psycopg2 "psycopg[binary]" asyncpg

2. Start MySQL and PostgreSQL services using Docker:

docker compose -f db.docker-compose.yml up

3. Set environment variables for database connections (or use defaults):

export TEST_URLS="mysql+mysqldb://test:test@127.0.0.1:3306/test mysql+pymysql://test:test@127.0.0.1:3306/test postgresql+psycopg2://postgres:test@127.0.0.1:5432/ postgresql+psycopg://postgres:test@127.0.0.1:5432/"
export TEST_ASYNC_URLS="mysql+aiomysql://test:test@127.0.0.1:3306/test mysql+asyncmy://test:test@127.0.0.1:3306/test postgresql+asyncpg://postgres:test@127.0.0.1:5432/ postgresql+psycopg://postgres:test@127.0.0.1:5432/"

ℹ️ Note: Test cases also load environment variables from tests/.env.

4. Run tests:

python -m unittest

Running Tests with Docker Compose

The project includes a comprehensive test matrix across Python and SQLAlchemy versions:

cd tests
docker compose up --abort-on-container-exit

Documentation

Full API documentation is available at https://sqlalchemy-dlock.readthedocs.io/

Links

• Source Code: https://github.com/tanbro/sqlalchemy-dlock
• Issue Tracker: https://github.com/tanbro/sqlalchemy-dlock/issues
• PyPI: https://pypi.org/project/sqlalchemy-dlock/