Storage Reference

Last updated: 2026-05-07 06:53:50

Storage Reference¶

The storage interface provides tenant-isolated key-value storage backed by PostgreSQL JSONB. Data is automatically scoped to the current tenant, user, and pack.

Overview¶

from huitzo_sdk import command, Context

@command("storage-demo", namespace="demo")
async def storage_demo(args: Args, ctx: Context) -> dict:
    # Save data
    await ctx.storage.save("my-key", {"field": "value"})

    # Retrieve data
    data = await ctx.storage.get("my-key")

    return data

Automatic Scoping¶

All storage operations are automatically scoped. When you save data with key "my-key", it's actually stored as:

tenant:{tenant_id}:user:{user_id}:pack:{pack_id}:my-key

This means: - Different tenants can't see each other's data - Different users (in the same tenant) can have their own data - Different packs have isolated storage

Core Methods¶

save(key, value, **options)¶

Save a value to storage.

# Simple save
await ctx.storage.save("settings", {"theme": "dark"})

# With TTL (expires in 1 hour)
await ctx.storage.save("cache:data", data, ttl=3600)

# Overwrite existing
await ctx.storage.save("counter", 0)
await ctx.storage.save("counter", 1)  # Replaces the value

Parameters:

Parameter	Type	Description
`key`	`str`	Storage key (max 256 chars)
`value`	`Any`	JSON-serializable value
`ttl`	`int`	Time-to-live in seconds (optional)
`scope`	`str`	Scope level: `"user"` (default), `"tenant"`, `"pack"`

get(key, **options)¶

Retrieve a value from storage.

# Basic get
data = await ctx.storage.get("settings")

# With default value
data = await ctx.storage.get("settings", default={"theme": "light"})

# Returns None if not found (and no default)
data = await ctx.storage.get("nonexistent")  # None

Parameters:

Parameter	Type	Description
`key`	`str`	Storage key
`default`	`Any`	Default value if key not found

delete(key)¶

Delete a value from storage.

await ctx.storage.delete("temporary-data")

# No error if key doesn't exist
await ctx.storage.delete("nonexistent")

exists(key)¶

Check if a key exists.

if await ctx.storage.exists("user-preferences"):
    prefs = await ctx.storage.get("user-preferences")
else:
    prefs = default_preferences

list(prefix)¶

List keys matching a prefix.

# List all keys starting with "cache:"
keys = await ctx.storage.list(prefix="cache:")
# Returns: ["cache:users", "cache:products", "cache:orders"]

# List all keys (empty prefix)
all_keys = await ctx.storage.list()

Scope Levels¶

User Scope (Default)¶

Data is isolated per tenant + user + pack:

# Only this user in this tenant can see this data
await ctx.storage.save("my-settings", settings)
# Equivalent to:
await ctx.storage.save("my-settings", settings, scope="user")

Tenant Scope¶

Data is shared across all users and all packs in the tenant. There is no pack-level isolation — any pack installed in the same tenant reads and writes the same keys.

# All users and all packs in this tenant share this data
await ctx.storage.save("shared-config", config, scope="tenant")

# Good for: org-wide settings that every pack should read

Warning: Because scope='tenant' has no pack component, two different packs using the same key will silently overwrite each other's data. Use unique key prefixes (e.g. "my-pack:config") to avoid collisions, or prefer scope='pack' when cross-pack sharing is not required.

Pack Scope¶

Data is shared across all users in the tenant but isolated to this pack. This is the recommended scope when data should be shared among a pack's users without risk of collision with other packs.

# Shared within this pack across all users, isolated from other packs
await ctx.storage.save("pack-cache", cache, scope="pack")

# Good for: pack-level caches, computed lookups, pack-specific org settings

Data Types¶

Storage accepts any JSON-serializable value:

# Primitives
await ctx.storage.save("string", "hello")
await ctx.storage.save("number", 42)
await ctx.storage.save("float", 3.14)
await ctx.storage.save("boolean", True)
await ctx.storage.save("null", None)

# Collections
await ctx.storage.save("list", [1, 2, 3])
await ctx.storage.save("dict", {"key": "value"})

# Nested structures
await ctx.storage.save("complex", {
    "users": [
        {"name": "Alice", "age": 30},
        {"name": "Bob", "age": 25}
    ],
    "metadata": {
        "created": "2024-01-01",
        "version": 1
    }
})

# Pydantic models (auto-serialized)
from pydantic import BaseModel

class UserData(BaseModel):
    name: str
    email: str

user = UserData(name="Alice", email="[email protected]")
await ctx.storage.save("user", user.model_dump())

Key Naming Conventions¶

Use Prefixes for Organization¶

# Good: Clear prefixes
await ctx.storage.save("user:preferences", prefs)
await ctx.storage.save("cache:products", products)
await ctx.storage.save("report:2024-01", report)

# Bad: Flat keys
await ctx.storage.save("preferences", prefs)
await ctx.storage.save("products", products)

Use IDs in Keys¶

# Good: Include IDs for specific records
await ctx.storage.save(f"order:{order_id}", order_data)
await ctx.storage.save(f"analysis:{analysis_id}:result", result)

# Retrieve by ID
order = await ctx.storage.get(f"order:{order_id}")

Avoid Special Characters¶

# Good: Use colons and alphanumerics
await ctx.storage.save("user:settings:theme", "dark")

# Bad: Spaces, slashes, etc.
await ctx.storage.save("user settings/theme", "dark")  # Don't do this

TTL (Time-to-Live)¶

Set expiration for temporary data:

# Expires in 1 hour (3600 seconds)
await ctx.storage.save("session:token", token, ttl=3600)

# Expires in 1 day
await ctx.storage.save("cache:daily-report", report, ttl=86400)

# Expires in 5 minutes
await ctx.storage.save("rate-limit:user123", counter, ttl=300)

Note: Expired keys are cleaned up asynchronously. A get() on an expired key returns None.

Transactions¶

Perform multiple operations atomically:

# pseudocode — atomic transaction

async with ctx.storage.transaction():
    # All operations inside this block are atomic
    val_a = await ctx.storage.get(key_a)
    val_b = await ctx.storage.get(key_b)

    # Validate business rules
    if not valid(val_a, val_b):
        raise ValidationError(...)

    # Update both — committed together or rolled back together
    await ctx.storage.save(key_a, new_val_a)
    await ctx.storage.save(key_b, new_val_b)

If any operation in the transaction fails, all changes are rolled back.

Batch Operations¶

Efficiently work with multiple keys:

# pseudocode — batch operations

await ctx.storage.save_many({key1: val1, key2: val2, ...})
results = await ctx.storage.get_many([key1, key2, ...])   # returns dict, None for missing
await ctx.storage.delete_many([key1, key2, ...])

Query by Metadata¶

Add metadata for querying:

# Save with metadata
await ctx.storage.save(
    f"report:{report_id}",
    report_data,
    metadata={
        "type": "financial",
        "period": "2024-Q1",
        "status": "complete"
    }
)

# Query by metadata
reports = await ctx.storage.query(
    prefix="report:",
    metadata={"type": "financial", "status": "complete"}
)

Storage Limits¶

Limit	Value
Key length	256 characters
Value size	16 MB
Keys per tenant	100,000
Total storage per tenant	1 GB

Best Practices¶

1. Use Meaningful Keys¶

# ✅ Good: Descriptive, hierarchical
await ctx.storage.save(f"customer:{customer_id}:orders:latest", order)

# ❌ Bad: Cryptic, flat
await ctx.storage.save("o1", order)

2. Use TTL for Caches¶

# ✅ Good: Cache with expiration
await ctx.storage.save("cache:exchange-rates", rates, ttl=3600)

# ❌ Bad: Cache without expiration (accumulates forever)
await ctx.storage.save("cache:exchange-rates", rates)

3. Use Appropriate Scope¶

# ✅ Good: Shared data at tenant level
await ctx.storage.save("lookup:products", products, scope="tenant")

# ❌ Bad: Duplicating shared data per user
await ctx.storage.save("lookup:products", products)  # Each user has a copy

4. Handle Missing Keys¶

# ✅ Good: Use defaults
settings = await ctx.storage.get("settings", default={})

# ❌ Bad: Assume key exists
settings = await ctx.storage.get("settings")
settings["theme"]  # KeyError if settings is None!