0.72.0

3tears Core

Three-tier caching library for Python applications. Provides collections (L1 SQLite -> L2 NATS KV -> L3 PostgreSQL, or a git working tree) with subscript access, entity proxy objects, and configurable flush strategies.

Architecture

L1 (SQLite, in-process, sync)  ->  L2 (NATS KV, shared, async)  ->  L3 (PostgreSQL, persistent, async)

• L1: In-memory SQLite via WAL mode. Sync access. Used by entity attribute reads and writes.
• L2: NATS KV shared cache. Async. Cross-pod consistency for multi-instance deployments.
• L3: PostgreSQL (or PostGIS, YugabyteDB, etc.), or a git working tree (threetears.core.backends.git: rows as files, every change set one commit). Async. Source of truth.

Reads promote up the stack (L3 miss -> L2 miss -> L1 hit on next access). Writes flow down (L1 -> L2 -> L3, with optional deferred flush).

Quick Start

1. Configure the Registry

from threetears.core.collections.registry import CollectionRegistry
from threetears.core.cache.sqlite import SQLiteBackend
from threetears.nats import Principal, kv_key_scope_for

# Create and configure
l1 = SQLiteBackend("my_app_cache")
l1.initialize(sa_metadata)  # SQLAlchemy metadata with your table definitions

registry = CollectionRegistry()
registry.configure(
    l1_backend=l1,          # SQLiteBackend instance
    l2_client=nats_client,  # NATS client (optional, None to skip L2)
    # REQUIRED wherever an l2_client is: every L2 key is written as
    # `{scope}.{table}.{body}`, and the scope is the principal this process
    # authenticates to NATS as, so one process cannot read another's cached
    # rows. `configure()` raises `L2ScopeNotConfiguredError` without it.
    # Derive it -- never a literal -- so it cannot drift from the NATS grant,
    # which is minted from the same call.
    kv_key_scope=kv_key_scope_for(Principal.AGENT_POD, agent_id=AGENT_ID),
    l3_pool=postgres_pool,  # asyncpg pool
)

Bind the shared {namespace}-collections bucket once at startup, BEFORE that configure() call -- one identity declares the bucket and every other process binds it:

from threetears.core.collections import bind_collections_bucket

await bind_collections_bucket(nats_client, component="my-pod")

2. Per-Collection Pool Overrides

Different collections can use different databases:

# Default: all collections use YugabyteDB
registry.configure(l3_pool=yugabyte_pool)

# Override: geo collection uses PostGIS
registry.configure()  # keep defaults
# When creating the collection, register with override:
geo_collection = GeoCollection(registry, config, nats_client, write_buffer)
registry.register(geo_collection, l3_pool=postgis_pool)

3. Define a Collection

from threetears.core.collections.base import BaseCollection
from threetears.core.entities.base import BaseEntity

class UserEntity(BaseEntity):
    primary_key_field = "user_id"

class UsersCollection(BaseCollection[UserEntity]):
    primary_key_column = "user_id"

    @property
    def table_name(self) -> str:
        return "users"

    @property
    def entity_class(self) -> type[UserEntity]:
        return UserEntity

    async def fetch_from_store(self, entity_id):
        row = await self.l3_pool.fetchrow(
            "SELECT * FROM users WHERE user_id = $1", entity_id
        )
        return dict(row) if row else None

    async def save_to_store(self, data, original_timestamp=None):
        # INSERT or UPDATE with optimistic locking
        ...

    async def delete_from_store(self, entity_id):
        await self.l3_pool.execute(
            "DELETE FROM users WHERE user_id = $1", entity_id
        )

    def serialize(self, data):
        return json.dumps(data, default=str).encode()

    def deserialize(self, data):
        return json.loads(data)

4. Create Collection Instances

from threetears.core.collections.flush import WriteBuffer

write_buffer = WriteBuffer()
users = UsersCollection(registry, config, nats_client, write_buffer)

The config parameter must satisfy the CoreConfig protocol:

class CoreConfig(Protocol):
    collection_flush: str           # "ALWAYS", "ON_CHECKPOINT", "ON_SCHEDULE", "ON_SHUTDOWN"
    collection_flush_interval: int  # seconds between scheduled flushes
    collection_flush_tables: str    # comma-separated table names eligible for deferred flush

Access Patterns

Subscript Access (sync, transparent pull-through)

Subscript access is the primary API. On L1 miss, data is transparently pulled through L2/L3 via a background event loop. No await needed, no ensure() required:

# Read entity -- pulls through L2/L3 automatically on L1 miss
entity = users[user_id]

# Read single field
name = users[user_id, "name_display"]

# Write single field (writes to L1, tracks for flush)
users[user_id, "name_display"] = "New Name"

# Write full entity data (writes dict to L1)
users[user_id] = {"user_id": user_id, "name_display": "New Name", ...}

# Check if entity is in L1 (does NOT pull through -- L1 only)
if user_id in users:
    entity = users[user_id]

__getitem__ raises KeyError only if the entity doesn't exist in any tier. The L1 fast path is ~microseconds; an L1 miss with pull-through adds ~50-200us bridge overhead plus the actual L2/L3 I/O time.

For hot-path code where you want to avoid the sync-async bridge overhead on first access, you can pre-warm L1:

await users.ensure(user_id)  # async: pre-warms L1
entity = users[user_id]       # guaranteed L1 hit, no bridge needed

Async Operations

# Three-tier read: L1 -> L2 -> L3, promotes on miss. Returns None if not found.
entity = await users.get(user_id)

# Create a new entity (not persisted until save)
entity = users.create({"user_id": uuid7(), "name_display": "Alice", ...})

# Save through three-tier write path (L3 -> L1 -> L2)
await users.save_entity(entity)
# Or via entity directly:
await entity.save()

# Reload from L3 (discards local changes)
await entity.reload()

# Delete from all tiers
await users.delete(user_id)

# Invalidate L1 + L2 (force next read to hit L3)
await users.invalidate_cache(user_id)

Entity Attribute Access

Entities are thin cache proxies. Field data lives in L1, not in the entity object.

entity = await users.get(user_id)

# Read (checks entity._changes first, then L1 cache)
print(entity.name_display)

# Write (writes to L1 + tracks change)
entity.name_display = "Updated Name"

# Check dirty state
entity.is_dirty  # True after modification
entity.is_new    # True if created via collection.create()

# Get all changes
entity.get_changes()  # {"name_display": "Updated Name"}

# Export full entity data from L1
entity.to_dict()

# Persist
await entity.save()

Flush Strategies

Controls when deferred writes reach L3 (PostgreSQL):

Only tables listed in collection_flush_tables are eligible for deferred writes. All other tables always write immediately regardless of strategy.

L1 Staleness

Nothing cached in L1 ages. A row stays until a write that reaches it is heard: the collection broadcast evicts it, and a table carrying a write generation is followed, so a missed broadcast drops the table instead of serving it (docs/epoch-task-06-collection-generations-by-default.md). A cache that needs a bound is put on the epoch system, not given a timer.

Optimistic Locking

Collections use date_updated for optimistic locking. When saving an existing entity, the save_to_store implementation should check:

UPDATE users SET ... WHERE user_id = $1 AND date_updated = $2

If rows_affected == 0 for an UPDATE, BaseCollection.save_entity() raises ConcurrentModificationError.

Synchronous Callers

3tears is async. A command line, a batch script or a worker thread with no event loop calls it through threetears.core.sync:

from threetears.core import sync

page = sync.run(lambda: client.get("/status"), timeout=30)   # a factory or a coroutine

Every call runs on one shared event loop in a daemon thread, so a resource one call opens (a NATS connection, an HTTP client) serves the next. Open it through sync.run and use it only through sync.run. Called from inside a running event loop, run raises RuntimeError: await the work there instead.

A lease held across synchronous work is renewed on that loop while the caller's thread works:

nats = sync.run(connect_nats)                           # opened on the shared loop
lease = KVLease(nats, bucket_name="leases")
with lease.hold_sync("set.june", ttl=60, renew_every=20, max_wait_seconds=600) as held:
    for step in steps:
        held.check()                                    # raises LeaseLost once the lease is lost
        step()

hold_sync takes hold's keywords and returns a SyncHeldLease: held and lost_reason as on HeldLease, lost as a threading.Event, and check().

Health Checks

TracedHttpClient.probe(path, *, method="GET", headers=None, timeout=None) -> bool asks once whether an upstream answers (a status below 400). It never retries, never touches the circuit breaker, never raises for a network failure, and logs only at DEBUG, so a machine that is often off can be checked every minute without an ERROR each time. request and get retry and log an exhausted call at ERROR, which is right for real traffic and wrong for a health check.

Subclassing Guide

BaseEntity: Set primary_key_field to your PK column name. Add computed properties as needed. Do NOT store data in instance attributes. All data lives in L1.

BaseCollection: Set primary_key_column. Implement the 5 abstract methods: fetch_from_store, save_to_store, delete_from_store, serialize, deserialize. Use self.l3_pool for database access. Add domain-specific query methods (e.g., find_by_email).