Testing

Testing DynamoDB code usually means moto, localstack, or Docker. pydynox has a simpler option: an in-memory backend. No setup, no extra dependencies, just fast tests.

Why use this?

Zero dependencies - Already included in pydynox. No extra packages to install.
No cold start impact - Only loads when you import it. Your Lambda stays fast.
Auto-registered fixtures - pytest finds them automatically. No conftest.py needed.
Fast - In-memory storage. No network, no Docker.

Key features

In-memory backend that mimics DynamoDB
Auto-registered pytest fixtures
Seed data support
Test isolation (each test starts fresh)
Works with all pydynox operations

Getting started

Basic usage

Just add pydynox_memory_backend to your test function. That's it.

Async (default)Sync

"""Basic usage of pydynox_memory_backend fixture (async - default)."""

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


@pytest.mark.asyncio
async def test_create_user(pydynox_memory_backend):
    """Test creating a user - no DynamoDB needed!"""
    user = User(pk="USER#1", name="John", age=30)
    await user.save()

    found = await User.get(pk="USER#1")
    assert found is not None
    assert found.name == "John"
    assert found.age == 30


@pytest.mark.asyncio
async def test_update_user(pydynox_memory_backend):
    """Test updating a user."""
    user = User(pk="USER#1", name="Jane")
    await user.save()

    await user.update(name="Janet", age=25)

    found = await User.get(pk="USER#1")
    assert found.name == "Janet"
    assert found.age == 25


@pytest.mark.asyncio
async def test_delete_user(pydynox_memory_backend):
    """Test deleting a user."""
    user = User(pk="USER#1", name="Bob")
    await user.save()

    await user.delete()

    assert await User.get(pk="USER#1") is None

"""Basic usage of pydynox_memory_backend fixture (sync)."""

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


def test_create_user(pydynox_memory_backend):
    """Test creating a user - no DynamoDB needed!"""
    user = User(pk="USER#1", name="John", age=30)
    user.sync_save()

    found = User.sync_get(pk="USER#1")
    assert found is not None
    assert found.name == "John"
    assert found.age == 30


def test_update_user(pydynox_memory_backend):
    """Test updating a user."""
    user = User(pk="USER#1", name="Jane")
    user.sync_save()

    user.sync_update(name="Janet", age=25)

    found = User.sync_get(pk="USER#1")
    assert found.name == "Janet"
    assert found.age == 25


def test_delete_user(pydynox_memory_backend):
    """Test deleting a user."""
    user = User(pk="USER#1", name="Bob")
    user.sync_save()

    user.sync_delete()

    assert User.sync_get(pk="USER#1") is None

The fixture is auto-discovered by pytest. No imports needed, no conftest.py setup.

How it works

When you use pydynox_memory_backend:

pydynox switches to an in-memory storage
All save(), get(), query(), etc. use this storage
After the test, storage is cleared
The original client is restored

Each test is isolated. Data from one test doesn't leak to another.

Direct client calls

The fixture replaces pydynox's default client. Models use that client automatically. When your code calls the client API directly, use the client exposed by the fixture:

direct_client_calls.py

"""Use the memory backend fixture for direct client calls."""


def test_batch_write(pydynox_memory_backend):
    """Test direct batch operations without connecting to DynamoDB."""
    # GIVEN the client exposed by the memory backend fixture
    client = pydynox_memory_backend.client

    # WHEN we batch write and read two items
    client.sync_batch_write(
        "users",
        put_items=[
            {"pk": "USER#1", "name": "Alice"},
            {"pk": "USER#2", "name": "Bob"},
        ],
    )

    items = client.sync_batch_get(
        "users",
        [{"pk": "USER#1"}, {"pk": "USER#2"}],
    )

    # THEN both items are returned from in-memory storage
    assert len(items) == 2

Don't create a new DynamoDBClient() in this test. A new client uses its own AWS configuration and does not share the fixture's in-memory storage. The fixture client exposes only the operations supported by the memory backend. Use Moto in server mode or localstack when you need the full client.

Fixtures

pydynox provides three fixtures:

Fixture	Use case
`pydynox_memory_backend`	Most tests - empty database per test
`pydynox_memory_backend_factory`	Tests that need custom seed data
`pydynox_memory_backend_seeded`	Tests that share common seed data

pydynox_memory_backend

The simplest option. Each test starts with empty tables.

import pytest

@pytest.mark.asyncio
async def test_create_user(pydynox_memory_backend):
    user = User(pk="USER#1", name="John")
    await user.save()
    assert await User.get(pk="USER#1") is not None

pydynox_memory_backend_factory

Use when you need pre-populated data:

seed_data.py

"""Using seed data with pydynox_memory_backend_factory (async - default)."""

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


class Order(Model):
    model_config = ModelConfig(table="orders")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    total = NumberAttribute()


@pytest.mark.asyncio
async def test_with_seed_data(pydynox_memory_backend_factory):
    """Test with pre-populated data."""
    seed = {
        "users": [
            {"pk": "USER#1", "name": "Alice", "age": 30},
            {"pk": "USER#2", "name": "Bob", "age": 25},
        ]
    }

    with pydynox_memory_backend_factory(seed=seed):
        # Data is already there!
        alice = await User.get(pk="USER#1")
        assert alice.name == "Alice"

        bob = await User.get(pk="USER#2")
        assert bob.name == "Bob"


@pytest.mark.asyncio
async def test_with_multiple_tables(pydynox_memory_backend_factory):
    """Seed multiple tables at once."""
    seed = {
        "users": [{"pk": "USER#1", "name": "John", "age": 30}],
        "orders": [
            {"pk": "USER#1", "sk": "ORDER#001", "total": 100},
            {"pk": "USER#1", "sk": "ORDER#002", "total": 200},
        ],
    }

    with pydynox_memory_backend_factory(seed=seed):
        user = await User.get(pk="USER#1")
        assert user is not None

        orders = [order async for order in Order.query(partition_key="USER#1")]
        assert len(orders) == 2


# Alternative: use pydynox_seed fixture in conftest.py
@pytest.fixture
def pydynox_seed():
    """Override this in conftest.py to provide default seed data."""
    return {
        "users": [
            {"pk": "ADMIN#1", "name": "Admin", "age": 99},
        ]
    }


@pytest.mark.asyncio
async def test_with_seeded_fixture(pydynox_memory_backend_seeded):
    """Uses seed data from pydynox_seed fixture."""
    admin = await User.get(pk="ADMIN#1")
    assert admin.name == "Admin"

pydynox_memory_backend_seeded

For shared seed data across many tests, override pydynox_seed in your conftest.py:

# conftest.py
import pytest

@pytest.fixture
def pydynox_seed():
    return {
        "users": [
            {"pk": "ADMIN#1", "name": "Admin", "role": "admin"},
            {"pk": "USER#1", "name": "Test User", "role": "user"},
        ]
    }

Then use pydynox_memory_backend_seeded in your tests:

import pytest

@pytest.mark.asyncio
async def test_admin_exists(pydynox_memory_backend_seeded):
    admin = await User.get(pk="ADMIN#1")
    assert admin.role == "admin"

Inspecting data

Access the backend to inspect stored data:

inspect_data.py

"""Inspecting data in the memory backend (async - default)."""

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


class User(Model):
    model_config = ModelConfig(table="users")
    pk = StringAttribute(partition_key=True)
    name = StringAttribute()
    age = NumberAttribute(default=0)


@pytest.mark.asyncio
async def test_inspect_tables(pydynox_memory_backend):
    """Access the backend to inspect stored data."""
    await User(pk="USER#1", name="Alice").save()
    await User(pk="USER#2", name="Bob").save()

    # Access tables directly
    assert "users" in pydynox_memory_backend.tables
    assert len(pydynox_memory_backend.tables["users"]) == 2

    # Check specific items
    items = pydynox_memory_backend.tables["users"]
    pks = [item["pk"] for item in items.values()]
    assert "USER#1" in pks
    assert "USER#2" in pks


@pytest.mark.asyncio
async def test_clear_data(pydynox_memory_backend):
    """Clear data mid-test for fresh state."""
    await User(pk="USER#1", name="Test").save()
    assert await User.get(pk="USER#1") is not None

    # Clear all data
    pydynox_memory_backend.clear()

    # Now it's gone
    assert await User.get(pk="USER#1") is None


@pytest.mark.asyncio
async def test_isolation_between_tests(pydynox_memory_backend):
    """Each test starts with empty tables."""
    # This test doesn't see data from other tests
    assert await User.get(pk="USER#1") is None

    # Create data for this test only
    await User(pk="USER#1", name="Isolated").save()
    assert await User.get(pk="USER#1") is not None

Query and scan

The memory backend supports query and scan with filters:

query_scan.py

"""Testing query and scan operations (async - default)."""

import pytest
from pydynox import Model, ModelConfig
from pydynox.attributes import NumberAttribute, StringAttribute
from pydynox.conditions import Attr


class Order(Model):
    model_config = ModelConfig(table="orders")
    pk = StringAttribute(partition_key=True)
    sk = StringAttribute(sort_key=True)
    status = StringAttribute()
    total = NumberAttribute()


@pytest.mark.asyncio
async def test_query_by_partition_key(pydynox_memory_backend):
    """Test querying items by partition key."""
    await Order(pk="USER#1", sk="ORDER#001", status="pending", total=100).save()
    await Order(pk="USER#1", sk="ORDER#002", status="shipped", total=200).save()
    await Order(pk="USER#2", sk="ORDER#001", status="pending", total=50).save()

    # Query returns only USER#1's orders
    results = [order async for order in Order.query(partition_key="USER#1")]
    assert len(results) == 2


@pytest.mark.asyncio
async def test_query_with_filter(pydynox_memory_backend):
    """Test query with filter condition."""
    await Order(pk="USER#1", sk="ORDER#001", status="pending", total=100).save()
    await Order(pk="USER#1", sk="ORDER#002", status="shipped", total=200).save()
    await Order(pk="USER#1", sk="ORDER#003", status="pending", total=300).save()

    # Filter by status
    results = [
        order
        async for order in Order.query(
            partition_key="USER#1",
            filter_condition=Attr("status").eq("pending"),
        )
    ]
    assert len(results) == 2


@pytest.mark.asyncio
async def test_scan_all_items(pydynox_memory_backend):
    """Test scanning all items in a table."""
    await Order(pk="USER#1", sk="ORDER#001", status="pending", total=100).save()
    await Order(pk="USER#2", sk="ORDER#001", status="shipped", total=200).save()
    await Order(pk="USER#3", sk="ORDER#001", status="pending", total=300).save()

    results = [order async for order in Order.scan()]
    assert len(results) == 3


@pytest.mark.asyncio
async def test_scan_with_filter(pydynox_memory_backend):
    """Test scan with filter condition."""
    await Order(pk="USER#1", sk="ORDER#001", status="pending", total=100).save()
    await Order(pk="USER#2", sk="ORDER#001", status="shipped", total=200).save()
    await Order(pk="USER#3", sk="ORDER#001", status="pending", total=300).save()

    # Filter by total > 150
    results = [order async for order in Order.scan(filter_condition=Attr("total").gt(150))]
    assert len(results) == 2

Testing Lambda handlers

Perfect for testing AWS Lambda functions:

lambda_handler.py

"""Testing AWS Lambda handlers with pydynox."""

import asyncio

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


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


# Lambda handlers are sync, but can run async code inside
def create_user_handler(event, context):
    """Lambda handler that creates a user."""

    async def _create():
        user_id = event["user_id"]
        name = event["name"]

        user = User(pk=f"USER#{user_id}", name=name)
        await user.save()

        return {"statusCode": 201, "body": f"Created user {user_id}"}

    return asyncio.run(_create())


def get_user_handler(event, context):
    """Lambda handler that gets a user."""

    async def _get():
        user_id = event["user_id"]

        user = await User.get(pk=f"USER#{user_id}")
        if not user:
            return {"statusCode": 404, "body": "User not found"}

        return {"statusCode": 200, "body": {"name": user.name}}

    return asyncio.run(_get())


# Tests - no moto, no localstack, no DynamoDB Local!
def test_create_user_handler(pydynox_memory_backend):
    """Test the create user Lambda handler."""
    event = {"user_id": "123", "name": "John"}

    response = create_user_handler(event, None)

    assert response["statusCode"] == 201

    # Verify user was created (sync in test)
    user = User.sync_get(pk="USER#123")
    assert user is not None
    assert user.name == "John"


def test_get_user_handler(pydynox_memory_backend):
    """Test the get user Lambda handler."""
    # Setup: create a user first
    User(pk="USER#123", name="Jane").sync_save()

    event = {"user_id": "123"}
    response = get_user_handler(event, None)

    assert response["statusCode"] == 200
    assert response["body"]["name"] == "Jane"


def test_get_user_not_found(pydynox_memory_backend):
    """Test getting a non-existent user."""
    event = {"user_id": "999"}

    response = get_user_handler(event, None)

    assert response["statusCode"] == 404

No mocking needed. Your handler code runs exactly as it would in production, just with in-memory storage.

Without pytest

You can use MemoryBackend directly as a context manager or decorator:

context_manager.py

"""Using MemoryBackend as context manager (without pytest)."""

import asyncio

from pydynox import Model, ModelConfig
from pydynox.attributes import StringAttribute
from pydynox.testing import MemoryBackend


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


# Use as context manager (async)
async def main():
    with MemoryBackend() as backend:
        # All pydynox operations use in-memory storage
        user = User(pk="USER#1", name="John")
        await user.save()

        found = await User.get(pk="USER#1")
        print(f"Found: {found.name}")  # Output: Found: John

        # Inspect the data
        print(f"Tables: {list(backend.tables.keys())}")
        print(f"Items: {len(backend.tables['users'])}")


# Use with seed data (async)
async def test_with_seed():
    seed = {"users": [{"pk": "USER#1", "name": "Seeded"}]}

    with MemoryBackend(seed=seed):
        user = await User.get(pk="USER#1")
        assert user.name == "Seeded"
        print("Seed test passed!")


if __name__ == "__main__":
    asyncio.run(main())
    asyncio.run(test_with_seed())
    print("All tests passed!")

Supported operations

The memory backend supports:

Operation	Supported
`save()`	✓
`get()`	✓
`delete()`	✓
`update()`	✓
`query()`	✓
`scan()`	✓
`batch_write()`	✓
`batch_get()`	✓
Vector search	✓
Vector index lifecycle	✓
Conditions	✓
Atomic updates	✓
Distributed lock decorators and contexts	✓

Note

Some advanced features like transactions and GSI queries are not yet supported in the memory backend. Use localstack for those cases.

Vector search

The memory backend performs exact, deterministic vector search for cosine distance, Euclidean distance, and dot product. It also supports vector partition keys, equality-only inline filters, projections, and top_k.

with MemoryBackend():
    Product.sync_create_table()

    Product(
        pk="PRODUCT#1",
        tenant_id="TENANT#acme",
        embedding=[1.0, 0.0],
    ).sync_save()

    matches = Product.semantic.sync_search(
        [0.9, 0.1],
        partition_key="TENANT#acme",
        top_k=1,
    )

    assert matches[0].item.pk == "PRODUCT#1"

The memory backend uses exact search. Production vector indexes may not return the same ordering for vectors with nearly identical scores, so tests should avoid near-ties. Create the table or index in the test before searching; model declarations are not auto-discovered as live indexes.

Comparison with alternatives

Moto's mock_aws decorator patches boto3 and botocore clients. pydynox uses the AWS SDK for Rust, so the decorator does not intercept its requests. To use Moto with pydynox, run Moto in server mode and pass its URL as endpoint_url. For unit tests, prefer the built-in memory backend.

Feature	pydynox fixture	moto	localstack
Setup	None	Server mode + process	Docker
Speed	Fastest	Fast	Slow
Accuracy	Good	Good	Best
Dependencies	None	`moto[server]`	Docker
GSI support	No	Yes	Yes
Transactions	No	Yes	Yes

Use pydynox fixtures for:

Unit tests
Fast feedback loops
CI/CD pipelines
Simple CRUD tests

Use localstack for:

Integration tests
GSI queries
Transactions
Full DynamoDB compatibility

Tips

Run tests in parallel

The memory backend is isolated per test, so parallel execution works:

pytest -n auto  # with pytest-xdist

Combine with parametrize

import pytest

@pytest.mark.asyncio
@pytest.mark.parametrize("name,age", [
    ("Alice", 30),
    ("Bob", 25),
    ("Charlie", 35),
])
async def test_create_users(pydynox_memory_backend, name, age):
    user = User(pk=f"USER#{name}", name=name, age=age)
    await user.save()

    found = await User.get(pk=f"USER#{name}")
    assert found.age == age

Use autouse for all tests

If you want all tests to use the memory backend:

# conftest.py
import pytest

@pytest.fixture(autouse=True)
def use_memory_backend(pydynox_memory_backend):
    yield

Now every test automatically uses in-memory storage.

Next steps

Models - Define your data models
Query - Query items by key
Conditions - Conditional operations