Skip to content

tanbro/sqlalchemy-dlock

Repository files navigation

sqlalchemy-dlock

Python package PyPI Documentation Status codecov

sqlalchemy-dlock is a distributed-lock library based on Database and SQLAlchemy.

It currently supports below locks:

Database Lock
MySQL named lock
PostgreSQL advisory lock

Install

pip install sqlalchemy-dlock

Usage

  • Work with SQLAlchemy Connection:

    from sqlalchemy import create_engine
    from sqlalchemy_dlock import create_sadlock
    
    key = 'user/001'
    
    engine = create_engine('postgresql://scott:[email protected]/')
    conn = engine.connect()
    
    # Create the D-Lock on the connection
    lock = create_sadlock(conn, key)
    
    # it's not lock when constructed
    assert not lock.locked
    
    # lock
    lock.acquire()
    assert lock.locked
    
    # un-lock
    lock.release()
    assert not lock.locked
  • with statement

    from contextlib import closing
    
    from sqlalchemy import create_engine
    from sqlalchemy_dlock import create_sadlock
    
    key = 'user/001'
    
    engine = create_engine('postgresql://scott:[email protected]/')
    with engine.connect() as conn:
    
        # Create the D-Lock on the connection
        with create_sadlock(conn, key) as lock:
            # It's locked
            assert lock.locked
    
        # Auto un-locked
        assert not lock.locked
    
        # If do not want to be locked in `with`, a `closing` wrapper may help
        with closing(create_sadlock(conn, key)) as lock2:
            # It's NOT locked here !!!
            assert not lock2.locked
            # lock it now:
            lock2.acquire()
            assert lock2.locked
    
        # Auto un-locked
        assert not lock2.locked
  • Work with SQLAlchemy ORM Session:

    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy_dlock import create_sadlock
    
    key = 'user/001'
    
    engine = create_engine('postgresql://scott:[email protected]/')
    Session = sessionmaker(bind=engine)
    
    with Session() as session:
      with create_sadlock(session, key) as lock:
          assert lock.locked
      assert not lock.locked
  • Asynchronous I/O Support

    💡 TIP

    from sqlalchemy.ext.asyncio import create_async_engine
    from sqlalchemy_dlock.asyncio import create_async_sadlock
    
    key = 'user/001'
    
    engine = create_async_engine('postgresql+asyncpg://scott:[email protected]/')
    
    async with engine.connect() as conn:
        async with create_async_sadlock(conn, key) as lock:
            assert lock.locked
            await lock.release()
            assert not lock.locked
            await lock.acquire()
        assert not lock.locked

    ℹ️ NOTE
    aiomysql, asyncpg and psycopg are tested asynchronous drivers.

    We can install it with asynchronous DB libraries:

    pip install SQLAlchemy[asyncio] aiomysql sqlalchemy-dlock

    or

    pip install SQLAlchemy[asyncio] asyncpg sqlalchemy-dlock

Test

Following drivers are tested:

You can run unit-tests

  • on local environment:

    1. Install the project in editable mode with asyncio optional dependencies, and libraries/drivers needed in test. A virtual environment (venv) is strongly advised:

      pip install -e .[asyncio] -r tests/requirements.txt
    2. start up mysql and postgresql service

      There is a docker compose file db.docker-compose.yml in project's top directory, which can be used to run mysql and postgresql develop environment conveniently:

      docker compose -f db.docker-compose.yml up
    3. set environment variables TEST_URLS and TEST_ASYNC_URLS for sync and async database connection url. Multiple connections separated by space.

      eg: (following values are also the defaults, and can be omitted)

      TEST_URLS=mysql://test:[email protected]/test postgresql://postgres:[email protected]/
      TEST_ASYNC_URLS=mysql+aiomysql://test:[email protected]/test postgresql+asyncpg://postgres:[email protected]/

      ℹ️ NOTE
      The test cases would load environment variables from dot-env file tests/.env.

    4. run unit-test

      python -m unittest
  • or on docker compose:

    tests/docker-compose.yml defines a Python and SQLAlchemy version matrix -- it combines Python 3.8 to 3.12 and SQLAlchemy v1/v2 for test cases. We can run it by:

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