pydamodb 0.1.1

A Pydantic-based lightweight ODM for Amazon DynamoDB.

PydamoDB

PydamoDB is a lightweight Python library that gives your Pydantic models DynamoDB superpowers. If you're already using Pydantic for data validation and want a simple, intuitive way to persist your models to DynamoDB, this library is for you.

Features

• 🔄 Seamless Pydantic Integration - Your models remain valid Pydantic models with all their features intact.
• 🔑 Automatic Key Schema Detection - Reads partition/sort key configuration directly from your DynamoDB table.
• 📝 Conditional Writes - Support for conditional save, update, and delete operations.
• 🔍 Query Support - Query by partition key with sort key conditions and filters with built-in pagination.
• 🗂️ Index Support - Query Global Secondary Indexes (GSI) and Local Secondary Indexes (LSI).
• ⚠️ Rich Exception Hierarchy - Descriptive, catchable exceptions for all error cases.

Limitations

These are some limitations to be aware of:

• Float attributes: DynamoDB doesn't support floats. Use Decimal instead or a custom serializer.
• Key schema: Field names for partition/sort keys must match the table's key schema exactly.
• Transactions: Multi-item transactions are not supported.
• Scan operations: Full table scans are intentionally not exposed.
• Batch reads: Batch get operations are not supported.
• Update expressions: Only SET updates are supported. For ADD, REMOVE, or DELETE, read-modify-save the full item.

When to Use PydamoDB

This library IS for you if:

• You're already using Pydantic and want to persist models to DynamoDB.
• You want a simple, intuitive API without complex configuration.
• You prefer convention over configuration.

This library is NOT for you if:

• You need low-level DynamoDB control.
• You need a full-featured ODM (consider PynamoDB instead).
• You need complex multi-item transactions.

Installation

pip install pydamodb

Or with uv:

uv add pydamodb

Quick Start

Work with an instance

import boto3

from pydamodb import PrimaryKeyModel, PydamoConfig

dynamodb = boto3.resource("dynamodb")
characters_table = dynamodb.Table("characters")


class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)

    name: str           # Partition key
    age: int
    occupation: str
    catchphrase: str | None = None


# Create and save a character instance
homer = Character(name="Homer", age=39, occupation="Safety Inspector", catchphrase="D'oh!")
homer.save()

# Update the character
homer.age = 40
homer.save()

# Delete using the instance method
homer.delete()

Work with class methods

# Reuse the Character model defined above

# Retrieve a character
character = Character.get_item("Homer")
if character:
    print(f"Found {character.name}: {character.catchphrase}")

# Update a character
Character.update_item("Homer", updates={Character.attr.age: 40})

# Delete by key
Character.delete_item("Homer")

Core Concepts

Model Types

PydamoDB provides two base model classes:

PrimaryKeyModel (alias: PKModel)

Use for tables with only a partition key:

class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)

    name: str         # Partition key
    age: int
    occupation: str

Available methods:

• save() - Save the instance
• delete() - Delete the instance
• get_item(partition_key) - Get by partition key
• update_item(partition_key, updates=...) - Update by partition key
• delete_item(partition_key) - Delete by partition key

PrimaryKeyAndSortKeyModel (alias: PKSKModel)

Use for tables with both partition key and sort key. This is useful when you need to group related items and query them together:

class FamilyMember(PrimaryKeyAndSortKeyModel):
    pydamo_config = PydamoConfig(table=family_members_table)

    family: str       # Partition key
    name: str         # Sort key
    age: int
    occupation: str

Additional methods:

• query(partition_key, ...) - Query by partition key
• query_all(partition_key, ...) - Query all items (handles pagination)

Configuration

Each model requires a pydamo_config class variable with the DynamoDB table:

import boto3

from pydamodb import PrimaryKeyModel, PydamoConfig

dynamodb = boto3.resource("dynamodb")
characters_table = dynamodb.Table("characters")

class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)
    # ... fields

The table is a boto3 DynamoDB Table resource. PydamoDB automatically reads the key schema from the table.

CRUD Operations

Save

Save a model instance to DynamoDB:

homer = Character(name="Homer", age=39, occupation="Safety Inspector")
homer.save()

With conditions (optimistic locking, prevent overwrites):

from pydamodb import ConditionCheckFailedError

# Only save if the item doesn't exist
try:
    homer.save(condition=Character.attr.name.not_exists())
except ConditionCheckFailedError:
    print("Character already exists!")

# Only save if a field has a specific value
homer.save(condition=Character.attr.occupation == "Safety Inspector")

Get

Retrieve an item by its key:

# Partition key only table
character = Character.get_item("Homer")
if character is None:
    print("Character not found")

# With consistent read
character = Character.get_item("Homer", consistent_read=True)

For tables with partition key + sort key:

# Get a specific family member
member = FamilyMember.get_item("Simpson", "Homer")

Update

Update specific fields of an item:

# Update a single field
Character.update_item("Homer", updates={Character.attr.age: 40})

# Update multiple fields
Character.update_item("Homer", updates={
    Character.attr.age: 40,
    Character.attr.catchphrase: "Woo-hoo!",
})

# Conditional update
Character.update_item(
    "Homer",
    updates={Character.attr.occupation: "Astronaut"},
    condition=Character.attr.occupation == "Safety Inspector",
)

# Update the instance itself
homer = Character.get_item("Homer")
if homer:
    homer.age = 41
    homer.save()

For tables with partition key + sort key:

FamilyMember.update_item("Simpson", "Homer", updates={FamilyMember.attr.age: 40})

Delete

Delete an item:

# Delete instance
character = Character.get_item("Homer")
if character:
    character.delete()

# Delete by key
Character.delete_item("Homer")

# Conditional delete
Character.delete_item("Homer", condition=Character.attr.age > 50)

For tables with partition key + sort key:

FamilyMember.delete_item("Simpson", "Homer")

Query

Query items by partition key (only available for PrimaryKeyAndSortKeyModel):

# Get all members of a family
result = FamilyMember.query("Simpson")
for member in result.items:
    print(member.name, member.occupation)

# With sort key condition
result = FamilyMember.query(
    "Simpson",
    sort_key_condition=FamilyMember.attr.name.begins_with("B"),
)

# With filter condition
result = FamilyMember.query(
    "Simpson",
    filter_condition=FamilyMember.attr.age < 18,
)

# With limit
result = FamilyMember.query("Simpson", limit=2)

# Pagination
result = FamilyMember.query("Simpson")
while result.last_evaluated_key:
    result = FamilyMember.query(
        "Simpson",
        exclusive_start_key=result.last_evaluated_key,
    )
    # Process result.items

# Get all items (handles pagination automatically)
all_simpsons = FamilyMember.query_all("Simpson")

Batch write

PydamoDB wraps boto3's batch_writer so you can work directly with models.

characters = [
    Character(name="Homer", age=39, occupation="Safety Inspector"),
    Character(name="Marge", age=36, occupation="Homemaker"),
]

with Character.batch_writer() as writer:
    for character in characters:
        writer.put(character)

# Read them back with the usual helpers
homer = Character.get_item("Homer")
marge = Character.get_item("Marge")

Conditions

PydamoDB provides a rich set of condition expressions for conditional operations and query filters.

Comparison Conditions

# Equality
Character.attr.occupation == "Safety Inspector"   # Eq
Character.attr.occupation != "Teacher"            # Ne

# Numeric comparisons
Character.attr.age < 18               # Lt
Character.attr.age <= 39              # Lte
Character.attr.age > 10               # Gt
Character.attr.age >= 21              # Gte

# Between (inclusive)
Character.attr.age.between(10, 50)

Function Conditions

# String begins with
Character.attr.name.begins_with("B")

# Contains (for strings or sets)
Character.attr.catchphrase.contains("D'oh")

# IN - check if value is in a list
Character.attr.occupation.in_("Student", "Teacher", "Principal")
Character.attr.age.in_(10, 38, 39, 8, 1)

# Size - compare the size/length of an attribute
Character.attr.name.size() >= 3          # String length
Character.attr.children.size() > 0       # List item count
Character.attr.traits.size() == 5        # Set element count

# Attribute existence
Character.attr.catchphrase.exists()       # AttributeExists
Character.attr.retired_at.not_exists()    # AttributeNotExists

Logical Operators

Combine conditions using Python operators:

# AND - both conditions must be true
condition = (Character.attr.age >= 18) & (Character.attr.occupation == "Student")

# OR - either condition must be true
condition = (Character.attr.name == "Homer") | (Character.attr.name == "Marge")

# NOT - negate a condition
condition = ~(Character.attr.age < 18)

# Complex combinations
condition = (
    (Character.attr.age >= 10) &
    (Character.attr.occupation != "Baby") &
    ~(Character.attr.name == "Maggie")
)

Working with Indexes

Query Global Secondary Indexes (GSI) and Local Secondary Indexes (LSI):

class FamilyMember(PrimaryKeyAndSortKeyModel):
    pydamo_config = PydamoConfig(table=family_members_table)

    family: str         # Table partition key
    name: str           # Table sort key
    occupation: str     # GSI partition key (occupation-index)
    created_at: str     # LSI sort key (created-at-index)
    age: int


# Query a GSI (e.g., "occupation-index" with partition key "occupation")
inspectors = FamilyMember.query(
    partition_key_value="Safety Inspector",
    index_name="occupation-index",
)

# Query a LSI (same partition key as table, different sort key)
recent_simpsons = FamilyMember.query(
    partition_key_value="Simpson",
    sort_key_condition=FamilyMember.attr.created_at.begins_with("2024-"),
    index_name="created-at-index",
)

# Get all items from an index
all_students = FamilyMember.query_all(
    partition_key_value="Student",
    index_name="occupation-index",
)

Note: Consistent reads are not supported on Global Secondary Indexes.

Type-Safe Field Access

PydamoDB provides type-safe field access through the attr descriptor:

class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)

    name: str
    age: int
    occupation: str


# Type-safe field references
Character.attr.name        # ExpressionField[str]
Character.attr.age         # ExpressionField[int]

# Type checking catches errors
Character.update_item("Homer", updates={
    Character.attr.age: "not a number",  # Type error!
})

# Non-existent fields raise AttributeError
Character.attr.nonexistent  # AttributeError: 'Character' has no field 'nonexistent'

Mypy Plugin

For full type inference, enable the mypy plugin:

# pyproject.toml
[tool.mypy]
plugins = ["pydamodb.mypy"]

Error Handling

PydamoDB provides a rich exception hierarchy for precise error handling:

from pydamodb import (
    PydamoError,
    ConditionCheckFailedError,
    IndexNotFoundError,
    MissingSortKeyValueError,
)

# Catch all PydamoDB errors
try:
    homer.save()
except PydamoError as e:
    print(f"PydamoDB error: {e}")

# Catch specific errors
try:
    homer.save(condition=Character.attr.name.not_exists())
except ConditionCheckFailedError:
    print("Character already exists - condition failed")

try:
    FamilyMember.query("Simpson", index_name="nonexistent-index")
except IndexNotFoundError as e:
    print(f"Index not found: {e.index_name}")

try:
    FamilyMember.get_item("Simpson")  # Missing sort key!
except MissingSortKeyValueError:
    print("Sort key is required for this table")

Exception Hierarchy

PydamoError (base)
├── OperationError
│   ├── ConditionCheckFailedError
│   ├── MissingSortKeyValueError
│   ├── IndexNotFoundError
│   ├── TableNotFoundError
│   ├── ThroughputExceededError
│   └── DynamoDBClientError
└── ValidationError
    ├── InvalidKeySchemaError
    ├── InsufficientConditionsError
    ├── UnknownConditionTypeError
    └── EmptyUpdateError

Migrating from Pydantic

If you already have Pydantic models, migrating to PydamoDB is straightforward. Your models remain valid Pydantic models with all their features intact.

Step 1: Choose the Right Base Class

Your DynamoDB Table	Base Class to Use
Partition key only	PrimaryKeyModel
Partition key + Sort key	PrimaryKeyAndSortKeyModel

Step 2: Change the Base Class

# Before: Plain Pydantic model
from pydantic import BaseModel

class Character(BaseModel):
    name: str
    age: int
    occupation: str
    catchphrase: str | None = None


# After: PydamoDB model
from pydamodb import PrimaryKeyModel, PydamoConfig

class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)

    name: str           # Now serves as partition key
    age: int
    occupation: str
    catchphrase: str | None = None

Step 3: Match Field Names to Key Schema

Your model field names must match the attribute names in your DynamoDB table's key schema:

# If your table has partition key "name":
class Character(PrimaryKeyModel):
    name: str         # ✅ Must match partition key name exactly
    age: int          # Other fields can be named anything
    occupation: str

What Still Works

Everything you love about Pydantic continues to work:

class Character(PrimaryKeyModel):
    pydamo_config = PydamoConfig(table=characters_table)

    name: str
    age: int
    occupation: str
    catchphrase: str | None = None

    # ✅ Validators still work
    @field_validator("age")
    @classmethod
    def validate_age(cls, v: int) -> int:
        if v < 0:
            raise ValueError("Age cannot be negative")
        return v

    # ✅ Computed fields still work
    @computed_field
    @property
    def display_name(self) -> str:
        return f"{self.name} ({self.occupation})"


# ✅ model_dump() works
homer = Character(name="Homer", age=39, occupation="Safety Inspector")
data = homer.model_dump()

# ✅ model_validate() works
character = Character.model_validate({"name": "Homer", "age": 39, ...})

# ✅ JSON serialization works
json_str = homer.model_dump_json()

PydamoDB is designed to keep your models as valid Pydantic models. Anything that would break Pydantic functionality is avoided.

Migration Checklist

• Change base class from BaseModel to PrimaryKeyModel or PrimaryKeyAndSortKeyModel.
• Add pydamo_config = PydamoConfig(table=your_table) to the class.
• Ensure field names for keys match your DynamoDB table's key schema.

Contributing

Contributions are welcome! Please:

1. Fork the repository.
2. Create a feature branch.
3. Add tests for new functionality.
4. Submit a pull request.

Philosophy

PydamoDB is built on these principles:

• Simplicity over features: We don't implement every DynamoDB feature. The API should be intuitive and easy to learn.
• Pydantic-first: Your models should remain valid Pydantic models with all their features.
• Convention over configuration: Minimize boilerplate by reading configuration from your table.
• No magic: Operations do what they say. No hidden batch operations or automatic retries.

License

MIT License - see LICENSE file for details.