dyntastic 0.4.1

[](https://github.com/nayaverdier/dyntastic/actions/workflows/ci.yml)

dyntastic

A DynamoDB library on top of Pydantic and boto3.

Installation

pip3 install dyntastic

If the Pydantic binaries are too large for you (they can exceed 90MB), use the following:

pip3 uninstall pydantic  # if pydantic is already installed
pip3 install dyntastic --no-binary pydantic

Usage

The core functionality of this library is provided by the Dyntastic class.

Dyntastic is a subclass of Pydantic's BaseModel, so can be used in all the same places a Pydantic model can be used (FastAPI, etc).

import uuid
from datetime import datetime
from typing import Optional

from dyntastic import Dyntastic
from pydantic import Field

class Product(Dyntastic):
    __table_name__ = "products"
    __hash_key__ = "product_id"

    product_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None


class Event(Dyntastic):
    __table_name__ = "events"
    __hash_key__ = "event_id"
    __range_key__ = "timestamp"

    event_id: str
    timestamp: datetime
    data: dict

# All your favorite pydantic functionality still works:

p = Product(name="bread", price=3.49)
# Product(product_id='d2e91c30-e701-422f-b71b-465b02749f18', name='bread', description=None, price=3.49, tax=None)

p.dict()
# {'product_id': 'd2e91c30-e701-422f-b71b-465b02749f18', 'name': 'bread', 'description': None, 'price': 3.49, 'tax': None}

p.json()
# '{"product_id": "d2e91c30-e701-422f-b71b-465b02749f18", "name": "bread", "description": null, "price": 3.49, "tax": null}'

Inserting into DynamoDB

Using the Product example from above, simply:

product = Product(name="bread", description="Sourdough Bread", price=3.99)
product.product_id
# d2e91c30-e701-422f-b71b-465b02749f18

# Nothing is written to DynamoDB until .save() is called:
product.save()

Getting Items from DynamoDB

Product.get("d2e91c30-e701-422f-b71b-465b02749f18")
# Product(product_id='d2e91c30-e701-422f-b71b-465b02749f18', name='bread', description="Sourdough Bread", price=3.99, tax=None)

The range key must be provided if one is defined:

Event.get("d2e91c30-e701-422f-b71b-465b02749f18", "2022-02-12T18:27:55.837Z")

Consistent reads are supported:

Event.get(..., consistent_read=True)

A DoesNotExist error is raised by get if a key is not found:

Product.get("nonexistent")
# Traceback (most recent call last):
#   ...
# dyntastic.exceptions.DoesNotExist

Use safe_get instead to return None if the key is not found:

Product.safe_get("nonexistent")
# None

Querying Items in DynamoDB

# A is shorthand for the Attr class (i.e. attribute)
from dyntastic import A

# auto paging iterable
for event in Event.query("some_event_id"):
    print(event)


Event.query("some_event_id", per_page=10)
Event.query("some_event_id")
Event.query("some_event_id", range_key_condition=A.timestamp < datetime(2022, 2, 13))
Event.query("some_event_id", filter_condition=A.some_field == "foo")

# query an index
Event.query(A.my_other_field == 12345, index="my_other_field-index")

# note: Must provide a condition expression rather than just the value
Event.query(123545, index="my_other_field-index")  # errors!

# consistent read
Event.query("some_event_id", consistent_read=True)

If you need to manually handle pagination, use query_page:

page = Event.query_page(...)
page.items
# [...]
page.has_more
# True
page.last_evaluated_key
# {"event_id": "some_event_id", "timestamp": "..."}

Event.query_page(..., last_evaluated_key=page.last_evaluated_key)

Scanning Items in DynamoDB

Scanning is done identically to querying, except there are no hash key or range key conditions.

# auto paging iterable
for event in Event.scan():
    pass

Event.scan((A.my_field < 5) & (A.some_other_field.is_in(["a", "b", "c"])))
Event.scan(..., consistent_read=True)

Updating Items in DynamoDB

Examples:

my_item.update(A.my_field.set("new_value"))
my_item.update(A.my_field.set(A.another_field))
my_item.update(A.my_int.set(A.another_int - 10))
my_item.update(A.my_int.plus(1))
my_item.update(A.my_list.append("new_element"))
my_item.update(A.some_attribute.set_default("value_if_not_already_present"))

my_item.update(A.my_field.remove())
my_item.update(A.my_list.remove(2))  # remove by index

my_item.update(A.my_string_set.add("new_element"))
my_item.update(A.my_string_set.add({"new_1", "new_2"}))
my_item.update(A.my_string_set.delete("element_to_remove"))
my_item.update(A.my_string_set.delete({"remove_1", "remove_2"}))

The data is automatically refreshed after the update request. To disable this behavior, pass refresh=False:

my_item.update(..., refresh=False)

Supports conditions:

my_item.update(..., condition=A.my_field == "something")

By default, if the condition is not met, the update call will be a noop. To instead error in this situation, pass require_condition=True:

my_item.update(..., require_condition=True)

Changelog

0.4.1 2022-04-26

• Fixed serialization of dynamo types when using Pydantic aliases

0.4.0 2022-04-26

• Fixed compatibility with Pydantic aliases

0.3.0 2022-04-25

• Added support for nested attribute conditions and update expressions
• Fixed bug where refresh() would cause nested Pydantic models to be converted to dictionaries instead of loaded into their models
• Added Pydantic aliases (models will all be dumped using pydantic's by_alias=True flag).

0.2.0 2022-04-23

BREAKING: Accessing attributes after calling update(..., refresh=False) will trigger a ValueError. Read below for more information.

• Added built in safety for unrefreshed instances after an update. Any attribute accesses on an instance that was updated with refresh=False will raise a ValueError. This can be fixed by calling refresh() to get the most up-to-date data of the item, or by calling ignore_unrefreshed() to explicitly opt-in to using stale data.

0.1.0 2022-02-13

• Initial release