appendmuch 0.0.8

An extensible append-only log with in-memory cache and pluggable storage backends

appendmuch

An extensible append-only log with in-memory cache and pluggable storage backends. Ordering is based on a strict monotonic sequence counter, but timestamps are stored for informational purposes.

Appendmuch is intended for synchronous code or async applications that access a store from one event-loop thread. It is not safe for parallel processing: do not share a Store or driver across threads, processes, or concurrent workers without external serialization.

Installation

From PyPI:

pip install appendmuch

From GitHub:

pip install "appendmuch @ git+https://github.com/mrpg/appendmuch.git@master"

Quick start

from appendmuch import Memory, Sqlite3, Store

store = Store(Memory())  # or Store(Sqlite3("db.sqlite3"))

# Storage instances provide attribute-based access to namespaced data
player = store.storage("game", "player1")
player.score = 100
player.name = "Alice"

print(player.score)  # 100
print(player.name)   # Alice

Mutable values

Mutable values (lists, dicts, sets) require a context manager. Mutations are detected and persisted automatically on exit:

with player:
    player.items = ["sword"]
    player.items.append("shield")
# Changes are flushed here

with player:
    print(player.items)  # ['sword', 'shield']

Change tracking

A Store can call a custom function if changes are made to a Storage instance. This can be used to notify other parts of the software of internal state changes.

def update(ns, key, value):
    print(f"{key!r} on {ns!r} is now {value.data!r}")

store2 = Store(Memory(), on_change=update)
customer = store2.storage("firm", "customer")

customer.age = 37
customer.name_ = "John Doe"

Temporal queries with within

Every write is assigned a monotonically increasing sequence number (and a timestamp for reference). Use within to query field values as they were when a condition held:

from appendmuch import within

player.round = 1
player.score = 10
player.round = 2
player.score = 25

print(within(player, round=1).score)  # 10
print(within(player, round=2).score)  # 25

for round_val, ctx in within.along(player, "round"):
    print(f"Round {round_val}: score={ctx.score}")

By default, within returns the value in effect while the context holds, even if that value was set before the context field. Use within.strict when the queried field must have been written under that context:

player.treatment = "high"
player.round = 1
player.choice = "A"

print(within(player, round=1).treatment)         # high
print(within.strict(player, round=1).choice)     # A
print(within.strict(player, round=1).get("treatment"))  # None

for round_val, ctx in within.strict.along(player, "round"):
    print(f"Round {round_val}: choice={ctx.choice}")

Inspecting history

Changes are appended to the database, never overwritten, except for fields explicitly configured with replace semantics. The __history__() method on Storage instances returns a read-only mapping of field names to tuples of Values, a special validated type:

player.__history__()["score"]
# Returns:
# (Value(time=1771028451.3298147, unavailable=False, data=10, context='__main__.<module>:14'), ...)

A Value contains a context, indicating the approximate code location that triggered the change. Tombstones have unavailable=True. Each history tuple is sorted by seq (sequence number), not by time.

Virtual fields

Storage instances can be initialized with virtual fields that function similar to @propertys. This is a simple mechanism to enable more ORM-like behavior.

# Define helper
def get_group(player):
    return store.storage("game", player._group)

# Initialize Storage instances
player2 = store.storage("game", "player2", virtual={"group": get_group})
player3 = store.storage("game", "player3", virtual={"group": get_group})

# Note the underscore before "group"; this is accessed by get_group:
player2._group = player3._group = "group1"

# This is essentially get_group(player2).budget = 42.7:
player2.group.budget = 42.7

# Access from different player with same _group:
print(player3.group.budget)  # Also 42.7

Virtual fields can also be added and removed at any time using storage.virtual:

player = store.storage("game", "player1")
player.score = 100

# As a decorator:
@player.virtual
def score_doubled(p):
    return p.score * 2

print(player.score_doubled)  # 200

# With an explicit name:
@player.virtual("bonus")
def compute_bonus(p):
    return p.score * 0.1

# Or directly:
player.virtual["penalty"] = lambda p: p.score * -0.05

# Remove when no longer needed:
del player.virtual["penalty"]

References to Storage instances cannot be stored directly on Storage instances, but the following pattern helps with indirection:

def get_members(group):
    return [store.storage("game", p, virtual={"group": get_group}) for p in group._members]

def get_group(player):
    return store.storage("game", player._group, virtual={"members": get_members})

...

with player2.group:
    player2.group._members = ["player2", "player3"]

with player3.group as g:
    print(g.members)
    print(g.members[0].group.budget)  # Ha!

Custom types

The following types can be stored out-of-the-box: bool, int, float, str, tuple, bytes, complex, None, decimal.Decimal, frozenset, datetime.datetime, datetime.date, datetime.time, uuid.UUID, list, dict, bytearray, set, random.Random.

Note: orjson imposes some constraints on some particular values of some types. For example, math.inf is unavailable, and so are dicts with non-str keys. The same applies to certain uncommonly used subtypes of generics; for example, list[random.Random] is unavailable. An Exception will be raised if a value cannot be encoded, or if the encoded data does not decode back to the original value (as long as Codec.vigilant == True, which is the default).

Support for other types can be registered using a custom codec. Example. It would also be possible to write a codec that uses pickle, or similar, to handle more types.

Storage backends

• Memory: in-memory, ideal for testing
• Sqlite3: file-backed via SQLite (stdlib)
• PostgreSQL: PostgreSQL with connection pooling (requires psycopg)

Testing

pytest                                    # core tests (Memory driver)
pytest tests/test_driver_sqlite.py        # SQLite3 driver tests
APPENDMUCH_PG_CONNINFO="dbname=mydb" \
  pytest tests/test_driver_pg.py          # PostgreSQL driver tests

PostgreSQL tests are skipped automatically when APPENDMUCH_PG_CONNINFO is not set.

License

Everything in this repository is licensed under the GNU LGPL version 3.0, or, at your option, any later version. See LICENSE for details.

© Max R. P. Grossmann, Holger Gerhardt, 2026.