Persistence Guide¶

MAID provides a document store API for persisting game data. This guide covers how to define schemas, perform CRUD operations, and integrate persistence into your content packs.

Document Store Overview¶

The document store provides a simple, schema-based API for storing and retrieving data. It supports:

CRUD operations (Create, Read, Update, Delete)
Querying with filters and sorting
Multiple storage backends (PostgreSQL, in-memory)

Defining Schemas¶

Creating Data Models¶

Define your data models using Pydantic:

from datetime import datetime
from uuid import UUID

from pydantic import BaseModel, Field


class CharacterModel(BaseModel):
    """Persisted character data."""

    id: UUID
    name: str
    level: int = 1
    experience: int = 0
    class_name: str = "adventurer"
    race: str = "human"
    created_at: datetime = Field(default_factory=datetime.utcnow)
    last_login: datetime | None = None
    play_time_seconds: int = 0

    # Stats
    strength: int = 10
    dexterity: int = 10
    constitution: int = 10
    intelligence: int = 10
    wisdom: int = 10
    charisma: int = 10


class ItemModel(BaseModel):
    """Persisted item data."""

    id: UUID
    name: str
    description: str = ""
    item_type: str = "misc"
    weight: float = 0.0
    value: int = 0
    stackable: bool = False
    max_stack: int = 1


class RoomModel(BaseModel):
    """Persisted room data."""

    id: UUID
    name: str
    description: str
    zone_id: UUID | None = None
    exits: dict[str, UUID] = Field(default_factory=dict)
    flags: list[str] = Field(default_factory=list)

Registering Schemas¶

Register schemas in your content pack:

def register_document_schemas(self, store: DocumentStore) -> None:
    """Register document schemas for this pack."""
    store.register_schema("characters", CharacterModel)
    store.register_schema("items", ItemModel)
    store.register_schema("rooms", RoomModel)

Basic Operations¶

Getting a Collection¶

# Get a typed collection
characters = store.get_collection("characters")

Creating Documents¶

from uuid import uuid4

# Create a new character
character = CharacterModel(
    id=uuid4(),
    name="Aragorn",
    level=10,
    class_name="ranger",
)

# Store it
char_id = await characters.create(character)
print(f"Created character with ID: {char_id}")

# Or with a specific ID
specific_id = uuid4()
await characters.create(character, doc_id=specific_id)

Reading Documents¶

# Get by ID
character = await characters.get(char_id)
if character:
    print(f"Found: {character.name}")
else:
    print("Character not found")

# Get multiple by IDs
char_ids = [id1, id2, id3]
found_characters = await characters.get_many(char_ids)

Updating Documents¶

# Fetch, modify, and save
character = await characters.get(char_id)
if character:
    character.level += 1
    character.experience = 0
    updated = await characters.update(char_id, character)
    if updated:
        print("Character updated!")

Deleting Documents¶

deleted = await characters.delete(char_id)
if deleted:
    print("Character deleted")
else:
    print("Character not found")

Checking Existence¶

exists = await characters.exists(char_id)
if exists:
    print("Character exists")

Querying Documents¶

Query Options¶

Use QueryOptions to filter and sort results:

from maid_engine.storage.document_store import QueryOptions, SortOrder

options = QueryOptions(
    filters={"level": 10},        # Exact match filters
    order_by="name",              # Sort field
    order=SortOrder.ASC,          # Sort direction
    limit=20,                     # Max results
    offset=0,                     # Skip N results
)

results = await characters.query(options)

Filter Examples¶

# Find all level 10 characters
level_10 = await characters.query(QueryOptions(
    filters={"level": 10}
))

# Find all rangers
rangers = await characters.query(QueryOptions(
    filters={"class_name": "ranger"}
))

# Multiple filters (AND logic)
high_level_rangers = await characters.query(QueryOptions(
    filters={
        "class_name": "ranger",
        "level": 20,
    }
))

Sorting Results¶

# Sort by level descending
by_level = await characters.query(QueryOptions(
    order_by="level",
    order=SortOrder.DESC,
))

# Sort by name ascending
alphabetical = await characters.query(QueryOptions(
    order_by="name",
    order=SortOrder.ASC,
))

Pagination¶

# First page (items 1-20)
page1 = await characters.query(QueryOptions(
    order_by="name",
    limit=20,
    offset=0,
))

# Second page (items 21-40)
page2 = await characters.query(QueryOptions(
    order_by="name",
    limit=20,
    offset=20,
))

Counting Documents¶

# Count all
total = await characters.count()

# Count with filters
rangers_count = await characters.count({"class_name": "ranger"})

Storage Backends¶

In-Memory Store (Testing)¶

For testing, use the in-memory store:

from maid_engine.storage.document_store import InMemoryDocumentStore

store = InMemoryDocumentStore()
store.register_schema("characters", CharacterModel)

# Use normally - data is lost when the process ends

PostgreSQL Store (Production)¶

For production, use PostgreSQL with JSONB:

from maid_engine.storage.document_store import PostgresDocumentStore

store = PostgresDocumentStore(
    dsn="postgresql://user:pass@localhost/maid",
    table_name="documents",
    min_pool_size=2,
    max_pool_size=10,
)

# Initialize (creates tables)
await store.initialize()

# Register schemas
store.register_schema("characters", CharacterModel)

# Use the store...

# Clean up when done
await store.close()

Integration Patterns¶

In Content Packs¶

class MyContentPack(BaseContentPack):
    def register_document_schemas(self, store: DocumentStore) -> None:
        store.register_schema("my_data", MyDataModel)

    async def on_load(self, engine: GameEngine) -> None:
        # Access the store from the engine
        self._store = engine.document_store

        # Load initial data
        await self._load_world_data()

    async def _load_world_data(self) -> None:
        rooms = self._store.get_collection("rooms")
        all_rooms = await rooms.query(QueryOptions())
        for room in all_rooms:
            # Create room entities from persisted data
            pass

In Commands¶

async def save_command(ctx: CommandContext) -> bool:
    """Save current character to database."""
    if not ctx.document_store:
        await ctx.session.send("Database not available.")
        return True

    player = ctx.world.entities.get(ctx.player_id)
    if not player:
        return False

    # Convert entity to model
    character_data = entity_to_character_model(player)

    # Save
    characters = ctx.document_store.get_collection("characters")
    await characters.update(ctx.player_id, character_data)

    await ctx.session.send("Character saved!")
    return True

In Systems¶

class AutoSaveSystem(System):
    """Periodically saves player data."""

    priority = 200  # Run late
    _save_interval = 300.0  # 5 minutes
    _time_since_save = 0.0

    async def startup(self) -> None:
        self._store = self.world.document_store

    async def update(self, delta: float) -> None:
        self._time_since_save += delta

        if self._time_since_save >= self._save_interval:
            self._time_since_save = 0.0
            await self._save_all_players()

    async def _save_all_players(self) -> None:
        if not self._store:
            return

        characters = self._store.get_collection("characters")

        for entity in self.entities.with_tag("player"):
            if entity.has_tag("online"):
                model = entity_to_character_model(entity)
                await characters.update(entity.id, model)

Advanced Patterns¶

Caching¶

Cache frequently accessed data:

class RoomCache:
    def __init__(self, store: DocumentStore):
        self._store = store
        self._cache: dict[UUID, RoomModel] = {}

    async def get(self, room_id: UUID) -> RoomModel | None:
        if room_id in self._cache:
            return self._cache[room_id]

        rooms = self._store.get_collection("rooms")
        room = await rooms.get(room_id)
        if room:
            self._cache[room_id] = room
        return room

    def invalidate(self, room_id: UUID) -> None:
        self._cache.pop(room_id, None)

    def clear(self) -> None:
        self._cache.clear()

Transactions (PostgreSQL)¶

For complex operations, use database transactions:

async def transfer_item(
    store: PostgresDocumentStore,
    item_id: UUID,
    from_player_id: UUID,
    to_player_id: UUID,
) -> bool:
    """Transfer an item between players atomically."""
    # Note: This is a simplified example
    # Real implementation would use database transactions

    items = store.get_collection("items")
    item = await items.get(item_id)

    if not item or item.owner_id != from_player_id:
        return False

    item.owner_id = to_player_id
    return await items.update(item_id, item)

Migration Pattern¶

Handle schema changes gracefully:

class CharacterModelV2(BaseModel):
    """Updated character model with new fields."""

    id: UUID
    name: str
    level: int = 1
    # New field with default for migration
    gold: int = 0


async def migrate_characters(store: DocumentStore) -> int:
    """Migrate characters to new schema."""
    characters = store.get_collection("characters")
    all_chars = await characters.query(QueryOptions())

    migrated = 0
    for char_data in all_chars:
        # Old data won't have 'gold' field
        # Pydantic uses default value automatically
        updated = CharacterModelV2.model_validate(char_data.model_dump())
        await characters.update(char_data.id, updated)
        migrated += 1

    return migrated

Best Practices¶

Use Typed Collections¶

Always work with typed data:

# Good - typed
characters: DocumentCollection[CharacterModel] = store.get_collection("characters")
char = await characters.get(char_id)  # Returns CharacterModel | None

# The type system helps catch errors

Handle Missing Data¶

Always check for None returns:

character = await characters.get(char_id)
if character is None:
    # Handle missing data
    return

# Safe to use character now

Use Meaningful Collection Names¶

# Good - clear names
store.register_schema("player_characters", CharacterModel)
store.register_schema("world_rooms", RoomModel)
store.register_schema("item_templates", ItemTemplateModel)

# Avoid - ambiguous names
store.register_schema("data", SomeModel)
store.register_schema("stuff", OtherModel)

Batch Operations When Possible¶

# Good - single query
ids = [id1, id2, id3, id4, id5]
characters = await collection.get_many(ids)

# Avoid - multiple queries
characters = []
for id in ids:
    char = await collection.get(id)
    characters.append(char)

Consider Data Lifecycle¶

Define when data is created, updated, and deleted
Implement cleanup for orphaned data
Plan for data migration between versions

Next Steps¶

Testing Guide - Testing with the document store
Publishing Guide - Packaging your content pack
Configuration Reference - Database settings