Cache vs store vs db

When to use ctx.cache, ctx.store, or ctx.db — TTL semantics, scoping rules, and migration between layers

Looking for the LLM context layer?

This page covers ctx.cache, ctx.store, and ctx.db as extension-owned data surfaces — they are where your code reads and writes state. They are NOT the LLM's context channels. For how data reaches the LLM (skeleton, fact-ledger, classifier prompt), see Context channels. ctx.cache is the only surface on this page that crosses over — it can be used to back panel snapshots that handlers later expose to the LLM.

Every extension gets three persistence surfaces from the web-kernel: ctx.cache, ctx.store, and ctx.db. They are not interchangeable — each has a distinct lifetime, typing contract, and use case. Choosing the wrong one produces bugs that are subtle to diagnose (data that vanishes unexpectedly, bloated stores, unnecessary complexity). This page explains the differences, when to use each, and how to migrate between them.

The three persistence layers

	`ctx.cache`	`ctx.store`	`ctx.db`
Lifetime	TTL-bounded (5–300 seconds)	Permanent until deleted	Permanent (raw SQL)
Type contract	Pydantic `BaseModel` only	Free-form JSON (`dict`)	Raw SQL rows
Scoping	Per-user, per-extension, per-key	Per-user, per-extension, per-collection	Shared schema — no automatic isolation
Query model	Key lookup only	Collection + `where` filter	Arbitrary SQL
Registration	Requires `@ext.cache_model`	None	None
Max value size	64 KB per entry	No enforced cap	No enforced cap
Available in	All tool/panel/skeleton contexts	All tool/panel/skeleton contexts	All contexts
Use it for	Short-lived computed data, expensive fetches, panel-side view state	Extension state, user documents, credential storage, configuration	Analytical queries, complex joins, schemas the store abstraction does not fit

If you are not sure which layer to reach for, start with ctx.store. Most extensions never need ctx.db, and ctx.cache is an optimization added when store round-trips are measurably too slow or too frequent for a hot code path.

`ctx.cache`

Purpose

ctx.cache is a short-lived, Pydantic-typed key-value store that lives entirely within a single user's session. The web-kernel routes it through the Auth Gateway to a per-extension Redis namespace and enforces a strict TTL — entries cannot persist beyond 300 seconds. It is intentionally not a primary data store.

Use ctx.cache when:

You have computed a value that is expensive to recompute (an API response, an aggregated view), and you want to reuse it across multiple panel renders within the same user interaction.
Your skeleton tool precomputes data for a panel that will need the same data immediately after (write once, read in panel handler within the next few minutes).
You want a typed read-through pattern with automatic fallback to a live fetch.

Do not use ctx.cache when the data must outlive the current user session, or when it needs to be visible to a different user.

API

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class ResourceSummary(BaseModel):
    total: int
    active: int
    last_id: str


ext.cache_model("resource_summary")(ResourceSummary)


@ext.tool("get_resource_summary", description="Return resource summary for the current user.")
async def get_resource_summary(ctx) -> dict:
    cached = await ctx.cache.get("summary", ResourceSummary)
    if cached is not None:
        return {"status": "success", "data": cached.model_dump()}

    # Expensive operation — fetch live, then cache for 90 seconds
    total  = await ctx.store.count("resources")
    active = await ctx.store.count("resources", where={"is_active": True})
    page   = await ctx.store.query("resources", order_by="-updated_at", limit=1)
    last_id = page.data[0].id if page.data else ""

    summary = ResourceSummary(total=total, active=active, last_id=last_id)
    await ctx.cache.set("summary", summary, ttl_seconds=90)
    return {"status": "success", "data": summary.model_dump()}

`get_or_fetch` — the preferred read-through pattern

When a panel needs data that may or may not be cached, use get_or_fetch instead of a manual get + set:

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class DashboardView(BaseModel):
    spend_today: float
    clicks: int
    active_campaigns: int


ext.cache_model("dashboard_view")(DashboardView)


async def _fetch_dashboard(ctx) -> DashboardView:
    total  = await ctx.store.count("campaigns", where={"status": "active"})
    return DashboardView(spend_today=0.0, clicks=0, active_campaigns=total)


@ext.panel("dashboard")
async def panel_dashboard(ctx):
    data = await ctx.cache.get_or_fetch(
        "dashboard",
        DashboardView,
        fetcher=lambda: _fetch_dashboard(ctx),
        ttl_seconds=120,
    )
    # render data ...
    return {}

If the cache holds a valid entry, get_or_fetch returns it immediately. If not, it calls the fetcher, caches the result, and returns it — all in one call.

TTL semantics

TTL is expressed in integer seconds, clamped to [5, 300]. Values outside this range raise a ValueError at the SDK boundary before any network call. There is no way to store a cache entry permanently — that is intentional. If you find yourself wanting a longer TTL, the data belongs in ctx.store.

Key constraints

Cache keys must match [A-Za-z0-9_\-:] and must not exceed 128 characters. Namespacing with colons is common and encouraged:

# Good — stable, unambiguous, collision-free
email_slug: str = "user_at_example_com"
folder_slug: str = "INBOX"
key = f"inbox:{email_slug}:{folder_slug}"
key = "dashboard"
key = "summary:v2"

# Bad — spaces, slashes, and special characters are rejected
key = "my key"       # ValueError at SDK boundary
key = "a/b/c"        # ValueError at SDK boundary

Model registration

Every Pydantic class passed to ctx.cache.get or ctx.cache.set must first be registered with @ext.cache_model. Registration ties the class name to a stable string that appears in the Redis key and the cache envelope. If you call ctx.cache.get with an unregistered class, the SDK raises ValueError before any network request is made.

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class InboxPage(BaseModel):
    messages: list[dict]
    total: int


class UnreadSummary(BaseModel):
    count: int
    newest_from: str


# Register at module level — not inside a handler or function
ext.cache_model("inbox_page")(InboxPage)
ext.cache_model("unread_summary")(UnreadSummary)

Per-user scoping

ctx.cache is scoped to the current ctx.user.imperal_id and the current extension's app_id. Two users with the same key string are reading from completely separate namespaces. There is no cross-user cache access; the web-kernel enforces this at the HTTP layer.

Federal invariants

Invariant	Enforcement
`I-CACHE-TTL-CAP-300S`	`ttl_seconds` outside [5, 300] raises `ValueError` at SDK boundary
`I-CACHE-PYDANTIC-ONLY`	`set` rejects non-`BaseModel` values; `get` rejects non-`BaseModel` classes
`I-CACHE-MODEL-REGISTRATION-REQUIRED`	Unregistered model class raises `ValueError` before network
`I-CACHE-VALUE-SIZE-CAP-64KB`	Serialized envelope exceeding 64 KB raises `ValueError`
`I-CACHE-KEY-SAFETY`	Key character set and length enforced before network
`I-CACHE-GW-URL-DERIVE`	Context derives gateway URL from existing clients; raises `RuntimeError` in minimal mock contexts

`ctx.store`

Purpose

ctx.store is the primary persistence layer for extension data. Documents in the store are permanent — they are not evicted and do not expire. Think of it as a managed document database: each extension has its own partition, each user has their own namespace within that partition, and collections are the organizational unit.

Use ctx.store when:

Data must survive beyond the current user session (credentials, contacts, saved configurations, user-created documents).
You need to query by filter (where={"status": "active"}), sort by field, or count documents.
You need to fan out across all users from a system context (ctx.store.list_users).
The data is not just derived from other sources — it is the source of truth itself.

API overview

ctx.store exposes a small, consistent set of operations:

Method	Use
`create(collection, data)`	Insert a new document; returns `Document` with web-kernel-assigned `id`
`get(collection, doc_id)`	Fetch one document by ID; returns `Document
`query(collection, where=..., order_by=..., limit=...)`	Filtered collection scan; returns `Page[Document]`
`update(collection, doc_id, data)`	Replace document data by ID; returns updated `Document`
`delete(collection, doc_id)`	Remove document by ID; returns `bool`
`count(collection, where=...)`	Count documents matching a filter
`set(key, data)`	Upsert shortcut for `"collection/doc_id"` paths

Collection naming

Collection names must not contain : * ? [ ] whitespace /. Use underscores or hyphens to namespace by feature area. Names should be stable — once data is written to a collection, renaming it requires a migration.

# Good
"contacts"
"mail_accounts"
"wt_monitors"
"wt_scan_results"

# Bad — forbidden characters
"mail accounts"     # whitespace rejected
"monitors/sites"    # slash rejected

Common patterns

Create or update (upsert)

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.tool("save_contact", description="Save or update a contact.")
async def save_contact(ctx, email: str, name: str) -> dict:
    existing = await ctx.store.query("contacts", where={"email": email})
    if existing.data:
        doc = await ctx.store.update("contacts", existing.data[0].id, {"email": email, "name": name})
    else:
        doc = await ctx.store.create("contacts", {"email": email, "name": name})
    return {"status": "success", "id": doc.id}

Querying and filtering

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.tool("list_active_monitors", description="List all active monitors.")
async def list_active_monitors(ctx) -> dict:
    page = await ctx.store.query(
        "wt_monitors",
        where={"owner_id": ctx.user.imperal_id, "is_active": True},
        order_by="-updated_at",
        limit=50,
    )
    return {
        "status": "success",
        "items": [{"id": d.id, **d.data} for d in page.data],
    }

Fan-out from a scheduled handler

System-context handlers (e.g. @ext.schedule) cannot see user-owned documents directly. They must use list_users to discover which users have data, then switch into a per-user context via ctx.as_user:

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.schedule("daily_digest", cron="0 8 * * *")
async def send_daily_digest(ctx) -> None:
    async for user_ctx in ctx.store.list_users("contacts"):
        docs = await user_ctx.store.query("contacts")
        if docs.data:
            await user_ctx.notify(f"You have {len(docs.data)} contacts.")

See Scheduled tasks for the full system context pattern.

Per-user scoping

Store documents are scoped by the web-kernel to the current ctx.user.imperal_id and the current extension's app_id. A query("contacts") call from user A will never return documents that belong to user B. The platform enforces this at the Auth Gateway — extension code cannot access another user's store partition by any normal means.

list_users and query_all are system-context-only methods that deliberately cross this boundary under controlled conditions. Calling them from a non-system context raises RuntimeError.

`ctx.db`

Purpose

ctx.db is a raw database connection. It exposes acquire() and session() for direct SQL access. Most extensions never need it. The store abstraction handles the overwhelming majority of data access patterns, and it carries automatic per-user isolation, collection namespacing, and web-kernel-level audit trails.

Use ctx.db when:

You need analytical queries involving multiple collections that the store's where filter cannot express — for example, a JOIN across two collections with aggregate functions.
Your data schema is genuinely relational and the document model is a poor fit — for example, a time-series table with dozens of columns.
You need query predicates that the store's where interface does not support (range queries, NOT IN, subselects).

Do not use ctx.db as a shortcut to bypass store conventions, to write to tables owned by the platform, or to avoid the slightly higher-level store API. The store carries audit trail semantics and platform-enforced isolation that raw SQL bypasses.

Practical guidance

The db.acquire() path gives you a raw connection; db.session() gives you a transactional session. Both are async context managers. Because you are writing SQL directly, you are also responsible for the scoping and isolation that the store handles automatically:

from imperal_sdk import Extension

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


@ext.tool("fetch_spend_report", description="Return monthly spend totals per campaign.")
async def fetch_spend_report(ctx, year: int, month: int) -> dict:
    uid = ctx.user.imperal_id
    async with ctx.db.session() as session:
        rows = await session.execute(
            "SELECT campaign_id, SUM(spend) AS total FROM ad_spend "
            "WHERE user_id = :uid AND year = :year AND month = :month "
            "GROUP BY campaign_id ORDER BY total DESC LIMIT 50",
            {"uid": uid, "year": year, "month": month},
        )
    return {"status": "success", "rows": [dict(r) for r in rows]}

Note that the query explicitly filters by user_id. Unlike ctx.store, ctx.db does not inject any user isolation automatically.

When to avoid ctx.db:

Simple key-value access (a ctx.store.get is faster and safer).
CRUD operations on user documents (the store provides upsert, query, delete with isolation).
Anything that only involves a single collection (use ctx.store.query with where=).

Decision tree

Work through these questions in order:

1. Does the data need to survive beyond the next few minutes?

No → use ctx.cache with an appropriate TTL. Skip the remaining questions.
Yes → continue.

2. Is the value a derived view — computed from other data rather than a source of truth?

Yes, and it needs to be recomputed regularly → still consider ctx.cache if the TTL matches how stale the derived value can reasonably be.
No, this is primary state (credentials, user-created documents, extension configuration) → use ctx.store.

3. Can ctx.store.query with where= express your query?

Yes → use ctx.store.
No (you need joins, aggregates, range predicates, or cross-collection analytics) → consider ctx.db.

4. If you are considering ctx.db: does your extension actually own the schema?

Yes, and the data is genuinely relational → ctx.db is appropriate.
No, or the schema is owned by the platform → do not use ctx.db. File a ticket or restructure to use the store.

Common mistakes

Using cache for permanent data

async def _wrong_cache_for_credentials(ctx, creds) -> None:
    # WRONG — TTL expires, data vanishes
    await ctx.cache.set("user_credentials", creds, ttl_seconds=300)  # type: ignore[arg-type]


async def _right_store_for_credentials(ctx, creds) -> None:
    # RIGHT — credentials are permanent user state
    await ctx.store.create("credentials", creds.model_dump())

If a user's credentials disappear after five minutes and they have to reconnect, this is almost always a cache-as-store mistake.

Using store for short-lived computed values in hot paths

async def _wrong_store_for_ephemeral(ctx, snapshot_data, snapshot) -> None:
    # WRONG — creates and deletes documents on every panel render
    doc = await ctx.store.create("inbox_snapshot", snapshot_data)
    # ... use doc ...
    await ctx.store.delete("inbox_snapshot", doc.id)


async def _right_cache_for_ephemeral(ctx, snapshot) -> None:
    # RIGHT — cache is built for this
    await ctx.cache.set("inbox_snapshot", snapshot, ttl_seconds=60)  # type: ignore[arg-type]

Using the store for ephemeral computed state pollutes the collection, generates unnecessary writes, and leaves orphaned documents if the delete step is skipped on error.

Using db for simple key-value access

async def _wrong_db_for_kv(ctx) -> None:
    # WRONG — raw SQL for something the store handles natively
    async with ctx.db.session() as s:
        rows = await s.execute(
            "SELECT data FROM store_documents WHERE collection='config' AND user_id=:uid LIMIT 1",
            {"uid": ctx.user.imperal_id},
        )


async def _right_store_for_kv(ctx) -> None:
    # RIGHT — use the store
    doc = await ctx.store.get("config", "user_preferences")

Directly querying the store's underlying table bypasses all platform isolation guarantees, produces coupling to the internal schema, and will break if the schema changes.

Storing unregistered models in cache

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class InboxView(BaseModel):
    messages: list[dict]


# WRONG — InboxView is not registered, raises ValueError before network
async def _wrong_unregistered(ctx) -> None:
    await ctx.cache.set("inbox", InboxView(messages=[]), ttl_seconds=60)  # type: ignore[arg-type]


# RIGHT — register at module level before any handler runs
ext.cache_model("inbox_view")(InboxView)


async def _right_registered(ctx) -> None:
    await ctx.cache.set("inbox", InboxView(messages=[]), ttl_seconds=60)  # type: ignore[arg-type]

Forgetting user isolation in db queries

async def _wrong_no_user_filter(ctx) -> None:
    # WRONG — returns every user's rows
    async with ctx.db.session() as s:
        rows = await s.execute("SELECT * FROM ad_spend WHERE month = :month", {"month": 5})


async def _right_user_filter(ctx) -> None:
    # RIGHT — always filter by user
    uid = ctx.user.imperal_id
    async with ctx.db.session() as s:
        rows = await s.execute(
            "SELECT * FROM ad_spend WHERE user_id = :uid AND month = :month",
            {"uid": uid, "month": 5},
        )

Migration patterns

Promoting cache → store

When you realize a cached value is actually primary state (not a derived view), migrate it by writing to the store on the next cache miss and reading from the store thereafter. The cache can act as a read-through layer during a transition period.

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class ConnectionState(BaseModel):
    account_id: str
    connected: bool


ext.cache_model("connection_state")(ConnectionState)


async def get_connection(ctx) -> ConnectionState:
    # Check store first (primary source of truth after migration)
    doc = await ctx.store.get("connections", "primary")
    if doc is not None:
        return ConnectionState(**doc.data)

    # Fallback to cache during migration window
    cached = await ctx.cache.get("connection_state", ConnectionState)
    if cached is not None:
        # Promote to store so future reads skip the cache
        await ctx.store.create("connections", {"doc_id": "primary", **cached.model_dump()})
        return cached

    return ConnectionState(account_id="", connected=False)

Once all users have been migrated (their state has been written to the store at least once), remove the cache fallback path.

Demoting store → cache

When a store collection is being used for short-lived view state that is actually derived from other data, demote it to the cache. Delete the documents on the next read-path to clean up existing records.

from imperal_sdk import Extension
from pydantic import BaseModel

ext = Extension(
    "my-app",
    display_name="My App",
    description="My App extension — manage resources with AI assistance.",
    icon="icon.svg",
    actions_explicit=True,
)


class InboxSnapshot(BaseModel):
    message_ids: list[str]
    unread: int


ext.cache_model("inbox_snapshot")(InboxSnapshot)


async def get_inbox_snapshot(ctx) -> InboxSnapshot:
    # Try cache first (new path)
    cached = await ctx.cache.get("inbox_snapshot", InboxSnapshot)
    if cached is not None:
        return cached

    # Migrate: clean up legacy store document if it exists
    old = await ctx.store.get("inbox_snapshots", "latest")
    if old is not None:
        await ctx.store.delete("inbox_snapshots", old.id)

    # Build from live data and cache it
    snapshot = InboxSnapshot(message_ids=[], unread=0)  # simplified
    await ctx.cache.set("inbox_snapshot", snapshot, ttl_seconds=90)
    return snapshot

Introducing db

When you find yourself constructing complex queries from multiple ctx.store.query calls — fetching two collections and joining them in Python — consider whether a single ctx.db query would be cleaner. Before introducing ctx.db, verify that your extension owns the underlying schema and that the query cannot be expressed with store filters.

Federal invariants

Invariant	Layer	What it enforces
`I-CACHE-TTL-CAP-300S`	`ctx.cache`	TTL capped at 300 seconds — cache cannot masquerade as permanent storage
`I-CACHE-PYDANTIC-ONLY`	`ctx.cache`	Only Pydantic `BaseModel` instances accepted — no raw dicts or primitives
`I-CACHE-MODEL-REGISTRATION-REQUIRED`	`ctx.cache`	Model must be declared via `@ext.cache_model` before use
`I-CACHE-VALUE-SIZE-CAP-64KB`	`ctx.cache`	Serialized envelope capped at 64 KB per entry
`I-CACHE-KEY-SAFETY`	`ctx.cache`	Key character set `[A-Za-z0-9_\-:]`, max 128 chars
`I-STORE-THREAT-COUNTER-1`	`ctx.store`	Cross-context bypass attempts emit a SigNoz counter — any non-zero firing is an incident
`I-AS-USER-1`	`ctx.store`	`ctx.as_user()` requires system context; raises `RuntimeError` for non-system callers
`I-AS-USER-2`	`ctx.store`	Only `user.imperal_id` changes in `as_user()` — extension, tenant, and agency are preserved

Production examples

The following production extensions illustrate the layering pattern:

mail-client (mail-client/skeleton.py, mail-client/panels.py) — skeleton precomputes inbox page data and writes it to ctx.cache (InboxPage / UnreadSummary / InboxMessages models, TTL 90–120 seconds). Panel handlers read from cache via get_or_fetch. Persistent data (mail account credentials, contacts, last-read watermarks) lives in ctx.store collections (mail_accounts, contacts, mail_last_read). The cache is the panel's view layer; the store is the source of truth.

microsoft-ads (microsoft-ads/skeleton.py, microsoft-ads/panels.py) — skeleton_refresh_msads fetches live API data and writes a MsadsDashboard Pydantic model to ctx.cache with TTL 300 seconds. The left panel calls ctx.cache.get_or_fetch("dashboard", ...) — on a hit it renders immediately; on a miss it fetches live, caches the result, and returns it. OAuth credentials and account configuration live in ctx.store.

web-tools (web-tools/handlers_quick.py, web-tools/skeleton.py) — monitors, groups, snapshots, and scan results all live in ctx.store (wt_monitors, wt_snapshots, wt_scan_results, wt_ip_scan_results). The extension does not use ctx.cache because the data is primary state that must survive between sessions and must be queryable by filter.

Cache vs store vs db

The three persistence layers

`ctx.cache`

Purpose

API

`get_or_fetch` — the preferred read-through pattern

TTL semantics

Key constraints

Model registration

Per-user scoping

Federal invariants

`ctx.store`

Purpose

API overview

Collection naming

Common patterns

Per-user scoping

`ctx.db`

Purpose

Practical guidance

Decision tree

Common mistakes

Using cache for permanent data

Using store for short-lived computed values in hot paths

Using db for simple key-value access

Storing unregistered models in cache

Forgetting user isolation in db queries

Migration patterns

Promoting cache → store

Demoting store → cache

Introducing db

Federal invariants

Production examples

Cross-references

@ext.cache_model reference

Skeletons

Scheduled tasks

Web-kernel context

Federal invariants

On this page