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.

Production/Stable. v1.0.0 adds surgical per-signature invalidation, parameter-name exclusion from cache keys, store observability counters, PEP 561 typing markers, and HTTP helper parity - all on a frozen API surface under normal SemVer.

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 v1.0.0 Unlocks

1. Surgical invalidation - evict one call signature

Stop calling cache_clear() when only one entry needs to go. cache_invalidate() builds the exact key your function would use and deletes it.

@inhouse_cache(60)
def load_user(db: object, user_id: int) -> dict[str, int]:
    return {"user_id": user_id}

# After a write, evict only that user's entry
load_user.cache_invalidate(session, user_id=123)
# Returns True if a key was removed, False if it didn't exist

Same key_builder, exclude_types, and exclude closed over by the decorator. Match how the function is normally called - FastAPI routes typically need keywords (get_item.cache_invalidate(item_id=1)).

2. Exclude parameters from cache keys

Drop session objects, background tasks, or timestamps from keys without writing a custom key_builder.

@inhouse_cache(store=store, exclude=("db",))
def load_user(db: object, user_id: int) -> dict[str, int]:
    ...

# load_user(session_a, 1) and load_user(session_b, 1) share one cache entry

The hot path is unchanged when exclude is empty - the kwarg is not even passed to custom builders.

3. Store observability - hit ratios and eviction counters

MemoryStore.stats() returns hit, miss, eviction, set, and delete counters for logs, health checks, or Prometheus.

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

@inhouse_cache(store=store)
def load_item(item_id: int) -> dict:
    ...

print(store.stats())
# {'hits': 47, 'misses': 3, 'evictions': 0, 'sets': 3, 'deletes': 0, 'size': 3}

Hit ratio is left to callers: hits / (hits + misses) when the denominator is non-zero.

4. Typed package (PEP 561)

py.typed marker shipped in the wheel at inhouse/py.typed. MyPy and Pyright consume inline annotations - the package is officially typed.

5. HTTP helper parity on FastAPI

@fastapi_cache with http_cache=True now also exposes cache_clear() and cache_invalidate(). Previously only the plain mode had them.

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

get_item.cache_invalidate(item_id=1)  # works with http_cache=True
get_item.cache_clear()                # also works

6. Stable release - production/stable

Development Status moved to Production/Stable. API surface is frozen for the 1.x line under normal SemVer. No intentional key-format break from 0.3.0 when exclude is unused.

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, exclude=("db",))
async def load_user(db: object, 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, 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
Recursive key freezing - stable cache keys for sets, nested mappings, dataclasses, and Pydantic-like models
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() / entry_meta() - introspect seconds-until-expiry and stored ETag for a cached key
HTTP cache primitives in inhouse.http_cache - make_weak_etag, etag_matches, cache_control_header, http_cache_headers, http_cache_outcome (304 vs 200)
@inhouse_cache(etag=True) - store a stable weak ETag at write time (wire headers yourself or use a framework extra)
cache_clear() / cache_invalidate(*args, **kwargs) - clear all keys for a function, or one call signature
exclude=("db", ...) / exclude_types= / key_builder= - control what enters the cache key
MemoryStore.stats() - hit/miss/eviction/set/delete counters
py.typed - PEP 561 marker for MyPy/Pyright
disable_all() / INHOUSE_CACHE_DISABLE - global dev bypass
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 and return Starlette JSONResponse / 304 Response via 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 v1.0.0

cache_invalidate(*args, **kwargs) -> bool - every @inhouse_cache / @fastapi_cache wrapper exposes surgical key eviction. Builds the key with the same key_builder, exclude_types, and exclude closed over by the decorator, then store.delete(key)
exclude: Sequence[str] = () - parameter names omitted from key material. Arguments are bound via inspect.signature; named params in exclude drop from key material. Hot path unchanged when exclude is empty (kwarg not passed to custom builders)
MemoryStore.stats() -> dict[str, int] - observability counters: hits, misses, evictions, sets, deletes, size. Updated under the existing store lock
py.typed - PEP 561 marker shipped in the wheel. Strict MyPy already used in-repo; consumers now see the package as typed
HTTP helper parity - @fastapi_cache with http_cache=True now exposes cache_clear() and cache_invalidate() (previously only available in plain mode)
Production/Stable - Development Status classified as Production/Stable. API surface frozen for the 1.x line under normal SemVer. No intentional key-format break from 0.3.0 when exclude is unused

Migration from v0.3.0

Change	Action
Keys with default `exclude=()`	No intentional digest break from 0.3.0
New `exclude=(...)`	Opt-in; changes key shape for that decorator only
Custom `key_builder`	Must accept `exclude=` only if you pass a non-empty `exclude` to the decorator; empty `exclude` omits the kwarg (0.3.0 builders keep working)
FastAPI `http_cache=True`	Gains `cache_clear` / `cache_invalidate` - additive
Version / classifier	Bump dependency pin to `inhouse-cache>=1.0.0` if desired

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
`stats()`	Snapshot of hit/miss/eviction/set/delete counters plus current `size` (new in 1.0)

`@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
    exclude=(),                  # optional - parameter names 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(db: object, 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)`. Pass a custom callable for full control
`exclude_types`	`tuple[type, ...]`	`()`	Argument types excluded from key material (e.g. request objects)
`exclude`	`Sequence[str]`	`()`	Parameter names excluded from key material (e.g. `db`, `background_tasks`) (new in 1.0)
`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:

Helper	Description
`cache_clear()`	Remove all cached entries for this function
`cache_invalidate(args, *kwargs)`	Remove the entry for one call signature (same key rules as a normal call) (new in 1.0)

load_user.cache_invalidate(db_session, user_id=123)  # surgical
load_user.cache_clear()                               # all keys for load_user

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

Exclusion (1.0)

Mechanism	When to use
`exclude=("db",)`	Named params that must not affect the key
`exclude_types=(Session,)`	Drop by type (e.g. request objects)
`key_builder=...`	Domain-specific keys, composite IDs, etc.

`cache_clear()` and `cache_invalidate()`

Every @inhouse_cache-decorated function exposes .cache_clear() (all keys) and .cache_invalidate(*args, **kwargs) (one signature).

compile_rag_prompt.cache_clear()                          # after document ingestion
load_user.cache_invalidate(session, user_id=123)          # evict one entry
get_item.cache_invalidate(item_id=1)                      # FastAPI routes: keyword style

`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
`exclude_types`	`tuple[type, ...]`	`()`	Extra types omitted from key material (merged with Request/Response for the default builder)
`exclude`	`Sequence[str]`	`()`	Parameter names omitted from key material (new in 1.0)
`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

Decorated routes expose cache_clear() and cache_invalidate(*args, **kwargs) in all modes (HTTP parity is new in 1.0). Prefer keyword invalidation matching FastAPI call style.

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, routes return plain Python objects with 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.

HTTP Caching

In-process TTL covers the server. Optional HTTP layers cover the client/CDN. Core owns the semantics (http_cache_outcome); the FastAPI extra is a thin adapter that reads Request headers and returns Starlette Response objects. Other frameworks can wire the same helpers themselves.

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_invalidate(user_id=42)  # evict one user's orders after insert

`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	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.

cache_invalidate / cache_clear affect this process only in multi-worker setups.

Architecture

Core API

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

from inhouse import (
    CacheEntry,
    ExpirySweeper,
    HttpCacheOutcome,
    MemoryStore,
    cache,
    cache_control_header,
    caching_disabled,
    configure_default_store,
    disable_all,
    enable_all,
    etag_for_value,
    etag_matches,
    freeze_for_key,
    get_default_store,
    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 v1.0.0 Unlocks​

1. Surgical invalidation - evict one call signature​

2. Exclude parameters from cache keys​

3. Store observability - hit ratios and eviction counters​

4. Typed package (PEP 561)​

5. HTTP helper parity on FastAPI​

6. Stable release - production/stable​

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 v1.0.0​

Migration from v0.3.0​

Configuration Reference​

MemoryStore​

@inhouse_cache / cache()​

HTTP cache primitives (core - zero dependencies)​

Recursive key freezing​

Exclusion (1.0)​

cache_clear() and cache_invalidate()​

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​