Async support

pydynox supports async/await for high-concurrency applications like FastAPI, aiohttp, and other asyncio-based frameworks.

Why async?

Sync operations block the event loop:

async def handle_request(user_id: str):
    user = User.get(pk=user_id, sk="PROFILE")  # Blocks!
    # Other async tasks can't run while waiting for DynamoDB

Async operations let other tasks run while waiting for I/O:

async def handle_request(user_id: str):
    user = await User.async_get(pk=user_id, sk="PROFILE")  # Non-blocking
    # Other tasks can run while waiting

Model async methods

All Model CRUD operations have async versions with async_ prefix:

model_async.py

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    name = StringAttribute()
    age = NumberAttribute()


async def main():
    # Create and save
    user = User(pk="USER#123", sk="PROFILE", name="John", age=30)
    await user.async_save()

    # Get by key
    user = await User.async_get(pk="USER#123", sk="PROFILE")

    # Update
    await user.async_update(name="Jane", age=31)

    # Delete
    await user.async_delete()

Client async methods

The DynamoDBClient also has async versions:

client_async.py

from pydynox import DynamoDBClient


async def main():
    client = DynamoDBClient()

    # Get item
    await client.async_get_item("users", {"pk": "USER#123", "sk": "PROFILE"})

    # Put item
    await client.async_put_item("users", {"pk": "USER#123", "name": "John"})

    # Update item
    await client.async_update_item(
        "users",
        {"pk": "USER#123", "sk": "PROFILE"},
        updates={"name": "Jane"},
    )

    # Delete item
    await client.async_delete_item("users", {"pk": "USER#123", "sk": "PROFILE"})

Async query

Query returns an async iterator:

query_async.py

from pydynox import DynamoDBClient

client = DynamoDBClient()


async def iterate_results():
    # Iterate with async for
    async for item in client.async_query(
        "users",
        key_condition_expression="#pk = :pk",
        expression_attribute_names={"#pk": "pk"},
        expression_attribute_values={":pk": "USER#123"},
    ):
        print(item["name"])


async def collect_results():
    # Collect all results
    items = await client.async_query(
        "users",
        key_condition_expression="#pk = :pk",
        expression_attribute_names={"#pk": "pk"},
        expression_attribute_values={":pk": "USER#123"},
    ).to_list()
    return items

Concurrent operations

The real power of async is running operations concurrently:

concurrent.py

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    name = StringAttribute()
    age = NumberAttribute()


async def sequential():
    # Sequential - slow (100ms + 100ms + 100ms = 300ms)
    user1 = await User.async_get(pk="USER#1", sk="PROFILE")
    user2 = await User.async_get(pk="USER#2", sk="PROFILE")
    user3 = await User.async_get(pk="USER#3", sk="PROFILE")
    return user1, user2, user3


async def concurrent():
    # Concurrent - fast (~100ms total)
    user1, user2, user3 = await asyncio.gather(
        User.async_get(pk="USER#1", sk="PROFILE"),
        User.async_get(pk="USER#2", sk="PROFILE"),
        User.async_get(pk="USER#3", sk="PROFILE"),
    )
    return user1, user2, user3

Real world example

Fetch user and their orders at the same time:

real_world.py

import asyncio

from pydynox import DynamoDBClient, Model, ModelConfig
from pydynox.attributes import StringAttribute

client = DynamoDBClient()


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    name = StringAttribute()


async def get_user_with_orders(user_id: str):
    """Fetch user and their orders at the same time."""
    user, orders = await asyncio.gather(
        User.async_get(pk=f"USER#{user_id}", sk="PROFILE"),
        client.async_query(
            "orders",
            key_condition_expression="#pk = :pk",
            expression_attribute_names={"#pk": "pk"},
            expression_attribute_values={":pk": f"USER#{user_id}"},
        ).to_list(),
    )
    return {"user": user, "orders": orders}

FastAPI example

fastapi_example.py

from fastapi import FastAPI
from pydynox import DynamoDBClient, Model, ModelConfig, set_default_client
from pydynox.attributes import StringAttribute

app = FastAPI()
client = DynamoDBClient()
set_default_client(client)


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(hash_key=True)
    sk = StringAttribute(range_key=True)
    name = StringAttribute()


@app.get("/users/{user_id}")
async def get_user(user_id: str):
    user = await User.async_get(pk=f"USER#{user_id}", sk="PROFILE")
    if not user:
        return {"error": "User not found"}
    return {"name": user.name}


@app.post("/users/{user_id}")
async def create_user(user_id: str, name: str):
    user = User(pk=f"USER#{user_id}", sk="PROFILE", name=name)
    await user.async_save()
    return {"status": "created"}

Available async methods

Model

Sync	Async
`Model.get()`	`Model.async_get()`
`model.save()`	`model.async_save()`
`model.delete()`	`model.async_delete()`
`model.update()`	`model.async_update()`

DynamoDBClient

Sync	Async
`get_item()`	`async_get_item()`
`put_item()`	`async_put_item()`
`delete_item()`	`async_delete_item()`
`update_item()`	`async_update_item()`
`query()`	`async_query()`

Notes

Async methods use the same Rust core as sync methods
No extra dependencies needed
Works with any asyncio event loop
Hooks still run synchronously (before/after save, etc.)