SQLALCHEMY pessimisic/optimistic lock

pessimisic-optimistic_lock.py

import time

from sqlalchemy.orm import scoped_session

from src.utils import start_mapper

from sqlalchemy import engine, schema
from sqlalchemy.orm.session import Session, sessionmaker

from src.domain.posts.post_entity import PostEntity

from settings.dev import DevDataBaseConnection

start_mapper()


def get_engine():
    return engine.create_engine(
        DevDataBaseConnection.get_url(),
        pool_pre_ping=True,
        pool_recycle=3600,
        pool_size=10,
        max_overflow=5,
        pool_timeout=60,
        echo=True
    )


def get_session(sa_engine):
    _session = scoped_session(sessionmaker(
        bind=sa_engine,
        expire_on_commit=False,
        autocommit=False,
        autoflush=False
    ))

    return _session


def increase_like(session: Session):
    for _ in range(25):
        post = session.query(PostEntity).filter(PostEntity.pk == 1).one()
        post.like = 0

        session.commit()
        session.close()


def increase_like_with_pessimistic_lock(session):
    for _ in range(25):
        post = session.query(PostEntity).filter(PostEntity.pk == 1).with_for_update().one()
        post.like += 1
        session.commit()
        session.close()


import backoff


class OptimisticUpdateFailure(Exception):
    """ Transaction Faile """


# @backoff.on_exception(backoff.expo, OptimisticUpdateFailure, max_time=1)
def increase_like_by_optimistic_lock(session):
    for _ in range(25):
        post = session.query(PostEntity).filter(PostEntity.pk == 1).one()

        result = session.query(PostEntity).filter(
            PostEntity.pk == post.pk,
            PostEntity.version == post.version
        ).update({
            PostEntity.like: PostEntity.like + 1,
            PostEntity.version: PostEntity.version + 1
        })

        if not bool(result):
            session.rollback()
            session.close()
        else:
            session.commit()
            session.close()


def increase_like_by_optimistic_lock_with_sqlalchemy(session):
    for _ in range(25):
        post = session.query(PostEntity).filter(PostEntity.pk == 1).one()
        post.like = post.like + 1

        session.add(post)
        session.commit()
        session.close()


if __name__ == '__main__':
    import threading

    session = get_session(get_engine())

    t1 = threading.Thread(target=increase_like_by_optimistic_lock_with_sqlalchemy, args=(session,))
    t1.daemon = True
    t2 = threading.Thread(target=increase_like_by_optimistic_lock_with_sqlalchemy, args=(session,))
    t2.daemon = True

    t1.start()
    t2.start()

    t1.join()
    t2.join()

    post = session.query(PostEntity).filter(PostEntity.pk == 1).one()
    print("LIKE RESULT: ", post.like)
    print("VERSION: ", post.version)

    print("=" * 20)
    post = session.query(PostEntity).filter(PostEntity.pk == 1).one()
    post.like = 0
    post.version = 0

    session.commit()
    session.close()