Bloop is an object mapper for DynamoDB and DynamoDBStreams.

Requires Python 3.6+

pip install bloop

Usage

First, we need to import all the things:

>>> from bloop import (
...     BaseModel, Boolean, Column, String, UUID,
...     GlobalSecondaryIndex, Engine
... )

Next we'll define the account model (with streaming enabled), and create the backing table:

>>> class Account(BaseModel):
...     class Meta:
...         stream = {
...             "include": {"old", "new"}
...         }
...     id = Column(UUID, hash_key=True)
...     name = Column(String)
...     email = Column(String)
...     by_email = GlobalSecondaryIndex(projection='keys', hash_key='email')
...     verified = Column(Boolean, default=False)
...
>>> engine = Engine()
>>> engine.bind(Account)

Let's make a few users and persist them:

>>> import uuid
>>> admin = Account(id=uuid.uuid4(), email="[email protected]")
>>> admin.name = "Admin McAdminFace"
>>> support = Account(name="this-is-fine.jpg", email="[email protected]")
>>> support.id = uuid.uuid4()
>>> engine.save(admin, support)

Or do the same in a transaction:

>>> with engine.transaction() as tx:
...     tx.save(admin)
...     tx.save(support)
...
>>>

And find them again:

>>> q = engine.query(
...     Account.by_email,
...     key=Account.email=="[email protected]"
... )
>>> q.first()
Account(email='[email protected]',
        id=UUID('d30e343f-f067-4fe5-bc5e-0b00cdeaf2ba'),
        verified=False)

>>> s = engine.scan(
...     Account,
...     filter=Account.name.begins_with("Admin")
... )
>>> s.one()
Account(email='[email protected]',
        id=UUID('08da44ac-5ff6-4f70-8a3f-b75cadb4dd79'),
        name='Admin McAdminFace',
        verified=False)

Let's find them in the stream:

>>> stream = engine.stream(Account, "trim_horizon")
>>> next(stream)
{'key': None,
 'meta': {'created_at': datetime.datetime(...),
  'event': {'id': 'cbb9a9b45eb0a98889b7da85913a5c65',
   'type': 'insert',
   'version': '1.1'},
  'sequence_number': '100000000000588052489'},
 'new': Account(
            email='[email protected]',
            id=UUID('d30e343f-...-0b00cdeaf2ba'),
            name='this-is-fine.jpg',
            verified=False),
 'old': None}
>>> next(stream)
{'key': None,
 'meta': {'created_at': datetime.datetime(...),
  'event': {'id': 'cbdfac5671ea38b99017c4b43a8808ce',
   'type': 'insert',
   'version': '1.1'},
  'sequence_number': '200000000000588052506'},
 'new': Account(
            email='[email protected]',
            id=UUID('08da44ac-...-b75cadb4dd79'),
            name='Admin McAdminFace',
            verified=False),
 'old': None}
>>> next(stream)
>>> next(stream)
>>>

What's Next

Check out the User Guide or Public API Reference to create your own nested types, overlapping models, set up cross-region replication in less than 20 lines, and more!