demo of using bloop's transaction class

tx_demo.py

# Three-table solution to https://stackoverflow.com/q/35825034

# Generally this is a terrible idea, but it demos transactions (see the `new_user` method) and
# provides guaranteed uniqueness over a triplet of attributes.

# You could instead use optimistic writes (but be ready to clean up dangling records!), such as:
#   (1) write Username if pk not exist, else fail
#   (2) write Email if pk not exist, else roll back (1) and fail
#   (3) write User if pk not exist, else roll back (1), (2) and fail
#   (4) update (1), (2), (3) with attributes from each other so they are no longer dangling

#   keep in mind that if the process crashes before (4) you'll still have records to clean up, so you need
#   a low-velocity scan in the background.  This is a great time to try out the Streams feature!
#     https://bloop.readthedocs.io/en/latest/user/streams.html#retrieve-records

# If you decide to use a single table (eg. just the User class) you can't guarantee uniqueness,
# since a GSI is eventually consistent.

from bloop import BaseModel, Column, Condition, Engine, String


# https://bloop.readthedocs.io/en/latest/user/patterns.html#generic-if-not-exist
def if_not_exist(obj):
    condition = Condition()
    for key in obj.Meta.keys:
        condition &= key.is_(None)
    return condition


class Username(BaseModel):
    username = Column(String, hash_key=True)
    email = Column(String)
    uid = Column(String)


class Email(BaseModel):
    username = Column(String)
    email = Column(String, hash_key=True)
    uid = Column(String)


class User(BaseModel):
    username = Column(String)
    email = Column(String)
    uid = Column(String, hash_key=True)


def fail_on(**kwargs):
    """helper to make a tx fail.

    usage:
        fail_on(email="TAKEN", uid="RESERVED")
        # raises TransactionCanceled
        new_user(email="TAKEN", username="unique", uid="also unique")
        new_user(email="also unique", username="unique", uid="RESERVED")
    """
    for x, cls in {"email": Email, "username": Username, "uid": User}.items():
        if x in kwargs:
            engine.save(cls(**kwargs))


def new_user(email, username, uid):
    objs = [
        cls(username=username, email=email, uid=uid)
        for cls in {Username, Email, User}
    ]

    with engine.transaction() as tx:
        for obj in objs:
            tx.save(obj, condition=if_not_exist(obj))


engine = Engine(table_name_template="tx.{table_name}")
engine.bind(BaseModel)

Here's a sample of using fail_on to prep an object in just the Email table:

In [1]: from bloop.exceptions import TransactionCanceled
In [2]: fail_on(email="TAKEN")
In [3]: try:
   ...:     new_user(email="TAKEN", username="numberoverzero", uid="123")
   ...: except TransactionCanceled:
   ...:     print("commit failed")
   ...:     
commit failed

Once you're done you can clean up the tx.{Email, User, Username} tables.