inhouse

Zero-dependency, in-process TTL cache for Python. One decorator, stampede-safe, LRU-bounded. For when Redis is a meeting you don't want to have, or when you need to avoid yet another deployment. Designed to be simple and effective without bloat or complexity for developers.

Keep hot data cheap to serve - v0.3.0 adds vertical caching patterns for SQLite, RAG, and file-backed prompts alongside the existing HTTP layer.

Designed for easy use with FastAPI applications. Although FastAPI integration is absolutely optional.

Install

The package is published on PyPI as inhouse-cache. Imports use inhouse (e.g. from inhouse import MemoryStore).

Core - same wheel, no new pip extras:

pip install inhouse-cache

With FastAPI helpers (fastapi_cache, lifespan sweeper):

pip install inhouse-cache[fastapi]

Big Wins - What v0.3.0 Unlocks

1. Cache SQLite query results in memory

Stop re-running the same SELECT queries on every request. Cache read-heavy query results in memory instead of using SQLite as a cache layer.

from inhouse.sqlite import query_store, rows_to_dicts
from inhouse import inhouse_cache

store = query_store(default_ttl=60)

@inhouse_cache(store=store)
def fetch_settings(user_id: int) -> dict | None:
    row = get_db().execute(
        "SELECT * FROM settings WHERE user_id = ?", (user_id,)
    ).fetchone()
    return rows_to_dicts(row) if row else None

copy_on_read and safe_copy convert sqlite3.Row objects to plain dicts on read, so callers cannot corrupt cached rows. Connection objects stay out of cache keys for safe, pure-signature query functions.

2. Cache RAG prompt compilation

Cache repeated RAG work - retrieved chunks, compiled prompts, or both in one step. Same query plus same source context means no redoing vector search or template assembly on every request.

from inhouse.rag import rag_cache

@rag_cache(ttl_seconds=600)
async def compile_prompt(
    user_query: str,
    filters: dict,
    corpus_version: str,
) -> str:
    context = await vector_search(user_query, filters)
    return f"Context:\n{context}\n\nQuestion: {user_query}"

# after document ingestion:
compile_prompt.cache_clear()

Pass a corpus_version argument (timestamp, hash, migration id) for automatic miss when your knowledge base updates. Or call .cache_clear() after an ingestion run.

3. Context versioning

Pass a version string or timestamp with the request so the cache naturally misses when content is stale.

@rag_cache(ttl_seconds=600)
async def build_prompt(query: str, kb_version: str) -> str:
    # kb_version changes when your knowledge base updates
    # different version = different cache key = automatic miss
    chunks = await vector_search(query)
    return assemble_prompt(chunks, query)

# after reindexing:
build_prompt.cache_clear()

Great for RAG pipelines, configuration reloads, or any data that changes on a schedule.

4. Watch file-backed prompts - auto-bust on disk change

Load hundreds of pre-defined AI skills at startup into cached objects without re-reading disk on every request. Cache markdown, txt, or any file with system prompts and skill files in memory with long TTLs. The lib watches files and invalidates on timestamp changes when you edit a prompt locally.

from inhouse.files import file_cache

@file_cache(ttl_seconds=3600, watch_files=["*.md"])
def load_skill(path: str) -> str:
    return open(path, encoding="utf-8").read()

On cache hit, os.path.getmtime checks each watched file. If a file changed on disk, the entry is invalidated and re-read on the next call. Less I/O, same files on disk.

5. Per-function invalidation - no more `store.clear()`

@inhouse_cache(60)
async def load_catalog(category: str) -> list[dict]:
    ...

load_catalog.cache_clear()  # wipes ALL entries for this function only

Internally uses store.delete_prefix() with the function's module-qualified key prefix. Other cached functions are untouched. Same ergonomics as functools.lru_cache.

6. Global dev bypass - disable cache in one line

from inhouse import disable_all, enable_all

disable_all()  # all @inhouse_cache calls skip the cache entirely

Or set INHOUSE_CACHE_DISABLE=1 before boot. Useful for prompt engineering, debugging, and benchmarking. No code changes needed.

7. Stable keys for sets, dataclasses, and Pydantic models

from inhouse import freeze_for_key, make_cache_key

# sets are order-independent
key1 = make_cache_key(search, ({"ai", "cache"},), {})
key2 = make_cache_key(search, ({"cache", "ai"}, {}))
assert key1 == key2  # True - deterministic regardless of set iteration order

freeze_for_key() recursively normalizes set, frozenset, list vs tuple, nested dict, dataclass, and Pydantic-like models - without JSON-serializing live objects.

8. Isolated FastAPI store - HTTP traffic won't evict your app cache

@fastapi_cache and @inhouse_cache now use separate default memory stores by default. Heavy HTTP route caching will not push out your app, RAG, or SQLite cache entries.

from inhouse import get_default_store
from inhouse.fastapi import get_fastapi_default_store

assert get_default_store() is not get_fastapi_default_store()
# different stores by default - share explicitly if you want a global budget

Quick Start Usage

Core (any Python project)

from inhouse import MemoryStore, inhouse_cache

store = MemoryStore(max_size=1024, default_ttl=60)

@inhouse_cache(store=store)
async def load_user(user_id: int) -> dict[str, int]:
    return {"user_id": user_id}

Works with both async def and def callables.

Core with HTTP cache metadata

from inhouse import MemoryStore, http_cache_outcome, inhouse_cache, make_cache_key

store = MemoryStore(max_size=1024, default_ttl=60)

@inhouse_cache(60, store=store, etag=True)
async def load_catalog(item_id: int) -> dict[str, int]:
    return {"item_id": item_id}

body = await load_catalog(1)
cache_key = make_cache_key(load_catalog, (1,), {})
outcome = http_cache_outcome(
    body,
    if_none_match=client_if_none_match,
    remaining_ttl=store.remaining_ttl(cache_key),
    stored_etag=store.get_etag(cache_key),
    http_cache=True,
    cache_visibility="public",
    use_etag=True,
)

Core with file watching

@inhouse_cache(3600, watch_files=["*.md"])
def load_prompt(path: str) -> str:
    return open(path, encoding="utf-8").read()

# file changes on disk → cache invalidated on next hit

SQLite Query Caching

import sqlite3
import threading
from inhouse import inhouse_cache
from inhouse.sqlite import query_store, rows_to_dicts

store = query_store(default_ttl=60)
_local = threading.local()

def get_db() -> sqlite3.Connection:
    if not getattr(_local, "conn", None):
        _local.conn = sqlite3.connect("app.db")
        _local.conn.row_factory = sqlite3.Row
    return _local.conn

@inhouse_cache(store=store)
def fetch_user_orders(user_id: int) -> list[dict]:
    rows = get_db().execute(
        "SELECT * FROM orders WHERE user_id = ? ORDER BY created_at DESC",
        (user_id,),
    ).fetchall()
    return rows_to_dicts(rows)

RAG Prompt Caching

from inhouse.rag import rag_cache

@rag_cache(ttl_seconds=600)
async def build_search_prompt(
    query: str,
    top_k: int = 5,
    filters: dict | None = None,
) -> str:
    chunks = await vector_search(query, top_k, filters or {})
    return "Context:\n" + "\n---\n".join(chunks) + f"\n\nQuestion: {query}"

File-backed AI Skills

from inhouse.files import file_cache

@file_cache(ttl_seconds=3600, watch_files=True)
def load_system_prompt(path: str) -> str:
    return open(path, encoding="utf-8").read()

FastAPI Use Case

import asyncio
from fastapi import FastAPI
from inhouse import MemoryStore
from inhouse.fastapi import create_lifespan, fastapi_cache

store = MemoryStore(max_size=1024, default_ttl=60)
app = FastAPI(lifespan=create_lifespan(store))

@app.get("/items/{item_id}")
@fastapi_cache(store=store)
async def get_item(item_id: int) -> dict[str, int]:
    await asyncio.sleep(0.1)
    return {"item_id": item_id}

Requires pip install inhouse-cache[fastapi].

FastAPI with HTTP Caching

@app.get("/catalog/{item_id}")
@fastapi_cache(
    60,
    store=store,
    sliding=True,
    http_cache=True,
    etag=True,
    cache_visibility="public",
)
async def get_catalog_item(item_id: int) -> dict[str, int]:
    await asyncio.sleep(0.1)
    return {"item_id": item_id}

Features

Core (zero dependencies)

TTL cache with lazy expiry on read
Sliding TTL - opt-in touch-on-read extends active entry lifetimes; idle entries still expire
LRU eviction when max_size is exceeded
Per-key singleflight stampede guard - concurrent misses on the same key coalesce to one computation. Backend errors propagate to all waiters; client disconnect on the leader no longer aborts in-flight cache population for followers
Deterministic cache keys - canonical JSON serialization with type-qualified fallbacks for custom objects. Keyword argument order and Request subclasses don't cause spurious cache misses
Thread-safe store for sync and async callables
Fixed, store-default, or callable TTL on each cache write
Opt-in copy_on_read on MemoryStore - deep-copy cached values on read to prevent caller mutation from corrupting the cache
remaining_ttl() - introspect seconds-until-expiry for a cached key
HTTP cache primitives - framework-agnostic helpers in inhouse.http_cache: weak ETag generation (make_weak_etag), etag_matches, cache_control_header, http_cache_headers, and http_cache_outcome (304 vs 200 decision + header dict)
@inhouse_cache(etag=True) - store a stable weak ETag alongside cached values at write time (no HTTP response objects; wire headers yourself or use a framework extra)
Recursive key freezing - stable cache keys for sets, nested mappings, dataclasses, and Pydantic-like models without JSON-serializing live objects
cache_clear() - invalidate all cached entries for one decorated function
disable_all() / INHOUSE_CACHE_DISABLE - global dev bypass for prompt engineering and debugging
watch_files - lazy mtime checks invalidate cached file-backed content when prompts change on disk
Vertical packages - inhouse.sqlite, inhouse.rag, inhouse.files (always installed, opt-in import)

Optional FastAPI extra (pip install inhouse-cache[fastapi])

@fastapi_cache with Request/Response-aware cache keys
Automatic HTTP wiring - http_cache=True and/or etag=True read If-None-Match from the injected Request and return native Starlette JSONResponse / 304 Response using core http_cache_outcome
Background expiry sweeper via FastAPI lifespan helpers
Clean lifespan shutdown - background sweeper cancels without noisy tracebacks
Isolated default store - @fastapi_cache uses its own MemoryStore by default so HTTP traffic does not evict core application cache entries

Vertical packages

Ship in the default wheel - no pip extra required. Import only what you need:

Package	Import	Purpose
`inhouse.sqlite`	`from inhouse.sqlite import query_store, safe_copy`	SQLite query result caching with `copy_on_read` + `sqlite3.Row` fallback
`inhouse.rag`	`from inhouse.rag import rag_cache`	RAG prompt compilation preset (`ttl_seconds=600`)
`inhouse.files`	`from inhouse.files import file_cache`	Markdown/txt prompt caching with `watch_files` + long TTL

What's New in v0.3.0

Recursive key freezing (freeze_for_key()) - list → tagged tuple, tuple → tagged tuple (distinct from list), set/frozenset → order-independent frozenset of frozen children, dict/Mapping → sorted key-value tuples, dataclass → ("dataclass", qualname, field values…), Pydantic v1/v2 duck-type → ("pydantic", qualname, frozen dump) - no pydantic import required
cache_clear() - every @inhouse_cache-decorated function exposes .cache_clear() to wipe all its entries via store.delete_prefix()
delete_prefix(prefix) - MemoryStore method to bulk-delete all keys starting with a prefix
disable_all() / enable_all() / caching_disabled() / INHOUSE_CACHE_DISABLE - module-level switch checked on every wrapper entry. When active: no store.get, no store.set, no singleflight - straight through to the underlying function
watch_files - three modes: False (default), True (auto-discover file paths in arguments), ["*.md", "*.txt"] (filter by fnmatch glob on basename), explicit paths. On hit: os.path.getmtime checks each watched file; any missing file or mtime mismatch → store.delete(key), recompute on next call
MemoryStore(copy_fn=...) - pluggable custom copy function when copy_on_read=True. safe_copy() in inhouse.sqlite tries copy.deepcopy, falls back to sqlite3.Row → dict conversion on TypeError
inhouse.sqlite - query_store(), safe_copy(), rows_to_dicts(). Preset query_store(default_ttl=60) configures MemoryStore with copy_on_read=True + copy_fn=safe_copy
inhouse.rag - rag_cache() is a thin wrapper over @inhouse_cache(ttl_seconds=600). Pass corpus_version for automatic key change or call .cache_clear() after ingestion
inhouse.files - file_cache(), discover_paths(), snapshot_mtimes(), files_changed(). file_cache(ttl_seconds=3600, watch_files=["*.md"]) preset
FastAPI store isolation - @fastapi_cache uses get_fastapi_default_store() instead of the core get_default_store(). Pass explicit store= to share a global budget
Unified sync/async decorator - @inhouse_cache / cache() uses inspect.iscoroutinefunction to handle sync SQLite queries and async RAG compilation with the same decorator

Migration from v0.2.x

Change	Action
Cache key digests	Deploy; accept misses for set/tuple/dataclass/model args
FastAPI default store	Pass explicit `store=` to both decorators if you relied on implicit sharing
HTTP cache / ETag / sliding TTL	Unchanged from v0.2.x

Configuration Reference

`MemoryStore`

from inhouse import MemoryStore

store = MemoryStore(max_size=1024, default_ttl=60, sliding=False)

Parameter / attribute	Type	Default	Description
`max_size`	`int`	`1024`	Maximum number of entries before LRU eviction
`default_ttl`	`float \| None`	`None`	Default TTL in seconds for `store.set()` and decorators that omit `ttl_seconds`
`copy_on_read`	`bool`	`False`	When `True`, `get()` returns a copy of cached values so callers cannot mutate the store
`copy_fn`	`Callable[[Any], Any] \| None`	`None`	Custom copy function when `copy_on_read=True` (default: `copy.deepcopy`)
`sliding` (init)	`bool`	`False`	Store-wide default for touch-on-read TTL extension on `set()`
`sliding` (property)	`bool`	-	Read-only; current store-wide sliding default
`default_ttl` (property)	`float \| None`	-	Mutable at runtime; affects future writes only
`size`	`int` (read-only)	-	Current number of cached entries

Store methods:

Method	Description
`get(key, *, default=MISS)`	Return a cached value, or `default` on miss/expiry. Extends expiry on read when entry is sliding. Deep-copies when `copy_on_read=True`
`set(key, value, ttl_seconds=None, *, sliding=None, etag=None, watch_mtimes=None)`	Write a value; uses `default_ttl` when `ttl_seconds` is omitted. `sliding=None` inherits store default
`remaining_ttl(key)`	Seconds until expiry for a live entry, or `None` on miss/expired
`entry_meta(key)`	`(remaining_ttl, etag)` tuple for a live entry, or `None` - single lock hop for HTTP header assembly
`get_etag(key)`	Stored ETag for a live entry, or `None`
`delete(key)`	Remove one entry
`delete_prefix(prefix)`	Remove all keys starting with `prefix`
`clear()`	Remove all entries
`purge_expired()`	Proactively delete expired entries
`keys()`	List current cache keys

`@inhouse_cache` / `cache()`

Core decorator. Works with both async def and def callables.

from inhouse import MemoryStore, inhouse_cache, make_cache_key

store = MemoryStore(default_ttl=60)

@inhouse_cache(
    ttl_seconds=60,              # optional - see Dynamic TTL below
    store=store,                 # optional - defaults to a module-level store
    key_builder=make_cache_key,  # optional - custom cache key strategy
    exclude_types=(object,),     # optional - types omitted from key material
    sliding=False,               # optional - touch-on-read TTL extension
    etag=False,                  # optional - store weak ETag metadata at write time
    watch_files=False,           # optional - invalidate on watched file mtime changes
)
async def load_user(user_id: int) -> dict[str, int]:
    return {"user_id": user_id}

Parameter	Type	Default	Description
`ttl_seconds`	`float \| Callable[[], float] \| None`	`None`	TTL in seconds for each cache write. See Dynamic TTL.
`store`	`MemoryStore \| None`	module default	Cache instance to read/write
`key_builder`	`Callable[..., str]`	`make_cache_key`	Builds the cache key from function identity + arguments. Non-JSON-serializable arguments fall back to `module.qualname:str(value)`
`exclude_types`	`tuple[type, ...]`	`()`	Argument types excluded from key material (e.g. request objects)
`sliding`	`bool`	`False`	When `True`, each successful read extends expiry by the entry's stored TTL duration
`etag`	`bool`	`False`	When `True`, store a weak ETag (`W/"<sha256>"`) at write time via `make_weak_etag`. Retrieve with `store.get_etag(key)` or `store.entry_meta(key)`
`watch_files`	`bool \| list[str]`	`False`	`True` auto-discovers file paths in arguments; `["*.md"]` filters by glob; explicit paths are always watched

Decorated functions expose cache_clear() to remove all cached entries for that function.

Global cache bypass:

from inhouse import disable_all, enable_all, caching_disabled

disable_all()              # or set INHOUSE_CACHE_DISABLE=1
enable_all()

inhouse_cache is an alias for cache.

Global default store helpers:

from inhouse import configure_default_store, get_default_store

store = MemoryStore(default_ttl=120)
configure_default_store(store)

@inhouse_cache()  # uses the configured default store + its default_ttl
async def load_config() -> dict[str, str]:
    ...

HTTP cache primitives (core - zero dependencies)

Core owns HTTP semantics (ETag digests, If-None-Match matching, Cache-Control header values, 304 vs 200 decisions). It does not return framework response objects - adapters wire HttpCacheOutcome into your stack.

from inhouse import (
    HttpCacheOutcome,
    cache_control_header,
    etag_matches,
    http_cache_headers,
    http_cache_outcome,
    make_weak_etag,
)

tag = make_weak_etag({"id": 1})                         # W/"<sha256>"
etag_matches('W/"abc", W/"other"', 'W/"abc"')            # True
cache_control_header(30.5, visibility="public")           # "public, max-age=31"

outcome: HttpCacheOutcome = http_cache_outcome(
    body,
    if_none_match=client_if_none_match,
    remaining_ttl=42.0,
    stored_etag=tag,
    http_cache=True,
    cache_visibility="private",
    use_etag=True,
)
# outcome.status_code → 200 or 304
# outcome.headers     → {"Cache-Control": ..., "ETag": ...}
# outcome.body        → body on 200, None on 304

Use with MemoryStore.set(..., etag=...), @inhouse_cache(etag=True), or manual make_weak_etag at write time. Pair with store.remaining_ttl() / store.entry_meta() when assembling headers on cache hits.

Recursive key freezing

Arguments are frozen, then hashed via the existing canonical JSON + SHA-256 pipeline:

list → tagged tuple (__list__)
tuple → tagged tuple (__tuple__) - distinct from list
set / frozenset → frozenset of frozen children
dict / Mapping → sorted tuple of (str(key), frozen_value) pairs
dataclass instances → ("dataclass", qualname, field values…)
Pydantic v1/v2 duck-type → ("pydantic", qualname, frozen dump) - no pydantic import required
everything else → existing module.qualname:str(value) fallback

from inhouse import freeze_for_key, make_cache_key, inhouse_cache

@inhouse_cache(60)
async def search(filters: dict[str, set[str]]) -> list[dict]:
    ...

# {"tags": {"ai", "cache"}} and {"tags": {"cache", "ai"}} → same cache key

`cache_clear()` and `delete_prefix()`

Every @inhouse_cache-decorated function exposes .cache_clear() (same ergonomics as functools.lru_cache). Internally deletes keys prefixed with {module}.{qualname}:.

compile_rag_prompt.cache_clear()  # after document ingestion

store.delete_prefix("my_module.my_function:")

`watch_files` - three modes

Value	Behavior
`False`	default - no file watching
`True`	walk `args`/`kwargs`; collect `str` paths where `os.path.isfile
`[".md", ".txt"]`	discover paths in arguments, filter by `fnmatch` on basename
`["/abs/or/rel/path.md"]`	explicit static paths, even if not passed at call time

On write: snapshot_mtimes(paths) stored on CacheEntry.watch_mtimes. On hit: any missing file or mtime mismatch → store.delete(key), recompute.

@inhouse_cache(3600, watch_files=["*.md"])
def load_prompt(path: str) -> str:
    ...

Sliding TTL

Fixed TTL expires on an absolute deadline set at write time. Sliding TTL extends that deadline on every successful read by the entry's stored TTL duration - so frequently accessed data stays warm while idle data still expires naturally.

@inhouse_cache(60, store=store, sliding=True)
async def load_active_session(session_id: str) -> dict[str, str]:
    ...

Why use it: active user sessions, hot configuration, or any data accessed repeatedly within a window should stay cached without arbitrary mid-activity expiry.

Caveat: a continuously read key can live indefinitely until LRU eviction at max_size. Callable TTL is still evaluated on write only; sliding reuses the duration stored at the last write.

`@fastapi_cache` (requires `inhouse-cache[fastapi]`)

FastAPI-friendly wrapper around inhouse_cache. Automatically excludes Starlette Request and Response objects from cache keys. Uses an isolated default MemoryStore (see get_fastapi_default_store()).

from inhouse.fastapi import create_lifespan, fastapi_cache

store = MemoryStore(max_size=512, default_ttl=60)
app = FastAPI(lifespan=create_lifespan(store, sweep_interval=30.0))

@app.get("/items/{item_id}")
@fastapi_cache(store=store)
async def get_item(item_id: int) -> dict[str, int]:
    ...

Parameter	Type	Default	Description
`ttl_seconds`	`float \| Callable[[], float] \| None`	`None`	Same semantics as `@inhouse_cache`
`store`	`MemoryStore \| None`	module default	Cache instance to read/write
`key_builder`	`Callable[..., str] \| None`	`make_fastapi_cache_key`	Custom key builder. Defaults exclude Starlette `Request`/`Response`; override delegates that responsibility to you
`sliding`	`bool`	`False`	Touch-on-read TTL extension (same as `@inhouse_cache`)
`http_cache`	`bool`	`False`	Emit `Cache-Control` with `max-age` from remaining in-process TTL
`cache_visibility`	`"private" \| "public"`	`"private"`	Cache-Control visibility. Use `"public"` only for CDN/browser-shared assets
`etag`	`bool`	`False`	Generate stable ETag, handle `If-None-Match` / `304 Not Modified` via core `http_cache_outcome`, return Starlette responses

Custom key_builder functions replace the FastAPI-aware default. To keep Request/Response exclusion, delegate to make_fastapi_cache_key or pass your own exclude_types.

When http_cache=False and etag=False, behavior is identical to v0.1.x (returns plain Python objects, no HTTP headers). With only etag=True on @inhouse_cache (no FastAPI), values are cached with ETag metadata but no HTTP responses are emitted.

FastAPI-injected request is used only to read If-None-Match; it is stripped before your route handler runs and excluded from cache keys via make_fastapi_cache_key.

Isolated default store:

from inhouse import get_default_store, configure_default_store
from inhouse.fastapi import (
    get_fastapi_default_store,
    configure_fastapi_default_store,
    fastapi_cache,
)

assert get_default_store() is not get_fastapi_default_store()  # isolated by default

shared = MemoryStore(max_size=2048, default_ttl=60)
configure_default_store(shared)
configure_fastapi_default_store(shared)  # opt-in shared budget

HTTP Caching

v0.1.x optimized the server (in-process TTL, stampede guard, LRU). v0.2.0 added HTTP caching via @fastapi_cache; v0.2.1 moves the semantics into core so any framework can use them. The FastAPI extra remains a thin adapter that reads Request headers and returns Starlette Response objects.

Three complementary layers:

Layer	Mechanism	What it saves
1. In-process cache	`@inhouse_cache` / `@fastapi_cache` / `MemoryStore`	Server compute on repeat hits
2. Time-based HTTP cache	`http_cache=True` → `Cache-Control: max-age=N`	The round trip entirely while fresh
3. Conditional HTTP cache	`etag=True` → `If-None-Match` / `304`	The response payload when the round trip happens anyway

HTTP Cache-Control (http_cache=True)

Offloads execution load entirely. If a client browser or a CDN (like Cloudflare) sees a valid Cache-Control: public, max-age=30 header, they won't even send the request to your server - saving bandwidth and compute.

max-age is derived from store.remaining_ttl(key) on cache hits, so HTTP freshness tracks in-process TTL (including sliding extensions)
Defaults to Cache-Control: private, max-age=N - safe for user-specific responses
Opt in to cache_visibility="public" for CDN-shared public assets
Core helper: cache_control_header() / http_cache_headers()

ETag / 304 Not Modified (etag=True)

When a client already has the current version, inhouse returns 304 Not Modified with an empty body instead of re-serializing and re-transmitting the full response - a huge bandwidth win on repeat requests.

Stable weak ETag (W/"<sha256>") computed at cache-write time via canonical JSON digest (make_weak_etag)
@inhouse_cache(etag=True) stores the tag; @fastapi_cache(etag=True) also handles conditional requests automatically
If-None-Match handled on cache hits and misses (recompute + matching ETag still skips body transfer)
Pairs naturally with Cache-Control: the browser/CDN may skip the request entirely; if a conditional request arrives after expiry, 304 skips the payload
Core helper: etag_matches() / http_cache_outcome()

FastAPI return types: http_cache / etag modes return Starlette JSONResponse or 304 Response. Best suited to JSON-serializable dict/list/Pydantic returns. Routes returning custom Response subclasses should omit http_cache / etag.

Other frameworks: use core http_cache_outcome with your own request header reads and response types - same semantics, no FastAPI import required.

Lifespan / Background Cleanup (requires `inhouse-cache[fastapi]`)

from inhouse.fastapi import create_lifespan, inhouse_lifespan

# Option A: pass directly to FastAPI
app = FastAPI(lifespan=create_lifespan(store, sweep_interval=30.0))

# Option B: use inside your own lifespan
async with inhouse_lifespan(store, sweep_interval=30.0):
    ...

Parameter	Type	Default	Description
`store`	`MemoryStore`	required	Store to sweep for expired entries
`sweep_interval`	`float`	`30.0`	Seconds between background purge runs

Vertical Packages - Full Recipes

`inhouse.sqlite`

Cache query results in memory, not SQLite-as-store. Pure-signature query functions: obtain db from thread-local or pool, not as a decorated argument.

import threading
import sqlite3
from inhouse import inhouse_cache
from inhouse.sqlite import query_store, rows_to_dicts

store = query_store(default_ttl=60)
_local = threading.local()

def get_db() -> sqlite3.Connection:
    if not getattr(_local, "conn", None):
        _local.conn = sqlite3.connect("app.db")
        _local.conn.row_factory = sqlite3.Row
    return _local.conn

@inhouse_cache(store=store)
def fetch_settings(user_id: int) -> dict | None:
    row = get_db().execute(
        "SELECT * FROM settings WHERE user_id = ?", (user_id,)
    ).fetchone()
    return rows_to_dicts(row) if row else None

@inhouse_cache(store=store)
def fetch_recent_orders(user_id: int, limit: int = 10) -> list[dict]:
    rows = get_db().execute(
        "SELECT * FROM orders WHERE user_id = ? ORDER BY created_at DESC LIMIT ?",
        (user_id, limit),
    ).fetchall()
    return rows_to_dicts(rows)

fetch_recent_orders.cache_clear()  # after a new order is placed

`inhouse.rag`

Cache compiled prompt strings. Context versioning: pass corpus_version (timestamp, hash, migration id) as an argument for automatic miss on corpus change. Programmatic eviction: compile_prompt.cache_clear() after ingestion.

from inhouse.rag import rag_cache

@rag_cache(ttl_seconds=600)
async def compile_prompt(
    user_query: str,
    filters: dict,
    corpus_version: str,
) -> str:
    context = await vector_search(user_query, filters)
    return f"Context:\n{context}\n\nQuestion: {user_query}"

# after document ingestion:
compile_prompt.cache_clear()

`inhouse.files`

from inhouse.files import file_cache

@file_cache(ttl_seconds=3600, watch_files=["*.md"])
def load_skill(path: str) -> str:
    return open(path, encoding="utf-8").read()

Low-level helpers available for custom wiring:

from inhouse.files import discover_paths, snapshot_mtimes, files_changed

paths = discover_paths(["/prompts/skill1.md", "/prompts/skill2.md"])
snapshot = snapshot_mtimes(paths)
if files_changed(snapshot):
    # reload and re-cache
    ...

Dynamic TTL

TTL is resolved when a value is written to the cache (on a miss), not on every read. Changing TTL settings does not retroactively extend entries already stored.

With sliding TTL, successful reads extend expiry by the TTL duration stored at the last write (not by re-evaluating a callable TTL).

1. Fixed TTL (per route)

@inhouse_cache(60, store=store)
async def load_user(user_id: int) -> dict[str, int]:
    ...

Always expires 60 seconds after the value is cached (unless sliding=True extends it on read).

2. Store Default (mutable at runtime)

store = MemoryStore(default_ttl=60)

@inhouse_cache(store=store)
async def load_config() -> dict[str, str]:
    ...

# Later - affects future cache writes only
store.default_ttl = 300

Omitting ttl_seconds on the decorator uses store.default_ttl. If both are missing, inhouse raises ValueError.

store.default_ttl is safe to change at runtime from other threads; new writes pick up the updated value atomically.

3. Callable TTL (evaluated on each write)

settings = {"cache_ttl": 60}

@inhouse_cache(lambda: settings["cache_ttl"], store=store)
async def load_dashboard() -> dict[str, str]:
    ...

settings["cache_ttl"] = 300  # next cache miss uses 300 seconds

Useful for feature flags, config files, or environment-driven TTL without redeploying.

Priority Order

When a cache miss is written, TTL is resolved as:

Callable ttl_seconds() result, if a callable was passed
Fixed ttl_seconds float, if provided
store.default_ttl, if set
Otherwise → ValueError

When to Use inhouse

Scenario	inhouse	Redis	fastapi-cache2
Single-node (Worker) FastAPI prototype	Excellent	Overkill	Great
SQLite query result caching	Great	Overkill	N/A
RAG prompt compilation caching	Great	Overkill	N/A
File-backed AI skill prompts	Great	Overkill	N/A
Zero external infrastructure	Yes	No	Depends on backend
Distributed multi-instance cache	No	Yes	Yes (with Redis)
Decorator-first developer UX	Yes	No	Yes
Browser/CDN HTTP caching (v0.2.0)	Yes (opt-in)	No	Depends on backend

Important Limitations

inhouse is per-process memory. If you run uvicorn main:app --workers 4, each worker maintains its own independent cache. That keeps the design simple and avoids shared infrastructure. It is not a distributed cache.

HTTP cache independence: in-process TTL and HTTP Cache-Control / ETag freshness are related but not identical. Calling store.delete() or waiting for in-process expiry does not invalidate browser or CDN copies. Plan HTTP cache durations and invalidation accordingly.

watch_files: one getmtime per watched path per cache hit - ideal for prompt files; measure before using on hot HTTP routes.

Architecture

Core API

The core package has no runtime dependencies. Import from inhouse directly:

from inhouse import (
    HttpCacheOutcome,
    MemoryStore,
    cache_control_header,
    caching_disabled,
    configure_default_store,
    disable_all,
    enable_all,
    etag_matches,
    freeze_for_key,
    http_cache_headers,
    http_cache_outcome,
    inhouse_cache,
    make_cache_key,
    make_weak_etag,
)

See the Configuration Reference for full decorator and store options.

License

MIT

Install​

Big Wins - What v0.3.0 Unlocks​

1. Cache SQLite query results in memory​

2. Cache RAG prompt compilation​

3. Context versioning​

4. Watch file-backed prompts - auto-bust on disk change​

5. Per-function invalidation - no more store.clear()​

6. Global dev bypass - disable cache in one line​

7. Stable keys for sets, dataclasses, and Pydantic models​

8. Isolated FastAPI store - HTTP traffic won't evict your app cache​

Quick Start Usage​

Core (any Python project)​

Core with HTTP cache metadata​

Core with file watching​

SQLite Query Caching​

RAG Prompt Caching​

File-backed AI Skills​

FastAPI Use Case​

FastAPI with HTTP Caching​

Features​

Vertical packages​

What's New in v0.3.0​

Migration from v0.2.x​

Configuration Reference​

MemoryStore​

@inhouse_cache / cache()​

HTTP cache primitives (core - zero dependencies)​

Recursive key freezing​

cache_clear() and delete_prefix()​

watch_files - three modes​

Sliding TTL​

@fastapi_cache (requires inhouse-cache[fastapi])​

HTTP Caching​

Lifespan / Background Cleanup (requires inhouse-cache[fastapi])​

Vertical Packages - Full Recipes​

inhouse.sqlite​

inhouse.rag​

inhouse.files​

Dynamic TTL​

1. Fixed TTL (per route)​

2. Store Default (mutable at runtime)​

3. Callable TTL (evaluated on each write)​

Priority Order​

When to Use inhouse​

Important Limitations​

Architecture​

Core API​

License​