fluxds 0.1.0

Declare a runtime data structure like a config block; swap, compose, and let a cost engine pick the backend. Zero dependencies.

fluxds

Declare a runtime data structure like a config block — then swap it, compose it, and let a cost engine pick the backend. Zero dependencies. Its own model base (no Pydantic).

from fluxds import FluxModel, Order

class Task(FluxModel):
    id: int
    priority: int
    class flux:
        order_by = "priority"     # extract by this field
        order    = Order.MIN      # lowest first
        index    = ("id",)        # O(1) lookup by id

q = Task.collection()
q.push(Task(id=1, priority=5))
q.push(Task(id=2, priority=2))

q.pop()             # -> Task(id=2, priority=2)   lowest priority, no negation tricks
q[1]                # -> Task(id=1, priority=5)   O(1) lookup
q.swap(Order.MAX)   # O(1) — flip extraction order, never rebuilds

---

Why

You fetched some data (hundreds to a few million rows — a working set that fits in RAM). Now you need to keep it ordered, findable, capped, and fresh — and you'll want to change those rules later without rewriting your code. Today you hand-roll a heap plus a dict plus manual eviction, and rewrite it in every project. fluxds is that plumbing, declared once.

It is not a database and not a big-data tool. It's the in-memory working-set layer — the same role Pydantic plays for validation, fluxds plays for runtime data structures.

Install

pip install fluxds                                    # once published to PyPI
pip install git+https://github.com/Inlionden/fluxds.git   # straight from GitHub
# or, from a clone:
pip install -e .

Requires Python 3.9+. No runtime dependencies.

The idea in one table

A data structure is two cost vectors: time (Big-O + a kind: worst / amortized / expected) and space. Using it = calling operations. So a swap is never free — it's a trade, a SWOT. fluxds makes that trade explicit and even picks the structure for you.

Pillar	What it's for
Swap	Change your mind cheaply — flip min↔max in O(1), or migrate to a new backend with the cost reported first.
Compose	Features are keywords, not classes — stack lookup, membership, TTL, and size-cap onto one collection.
Cost engine	recommend(workload) picks the cheapest backend for your op-mix and refuses one that can't serve a required op.
Customizable	Nothing fits? Write ~25 lines (Backend + @register_backend) and it inherits indexing, swap, compose, and cost analysis.

What ships in v0.1

Piece	API
Declarative model	FluxModel + a flux: block (order_by, order, index, structure, bounded)
Collection	push · pop · peek · swap · delete · update · [key] · in · by · rank · top · around · ordered · swap_structure
Backends	HeapBackend (O(1) order swap), SortedBackend (skip list: O(log n) rank/top/around)
Compose	compose(backend, HashIndex, BloomFilter, ExpiryTTL, BoundedSize, key=...)
Cost engine	swot(a, b), recommend(workload), register_profile()
Rate limiting	SlidingWindow, TieredLimiter
Your own DS	Backend + @register_backend(name, costs=...), provides={...}

Examples

Leaderboard (rank / top-k / around)

class Entry(FluxModel):
    player: str
    score: int
    class flux:
        order_by = "score"; order = Order.MAX
        index = ("player",); structure = "sorted"

board = Entry.collection()
board.push(Entry(player="alice", score=1200))
board.update("alice", score=1450)   # re-ranks automatically
board.top(10)                        # the podium, non-destructive
board.rank("alice")                  # 1-based position
board.around("alice", 2)             # neighbours by rank

A cache from composed layers

from fluxds import compose, HeapBackend, HashIndex, BloomFilter, ExpiryTTL, BoundedSize

cache = compose(
    HeapBackend(order="min"),                          # ordering (oldest first)
    HashIndex("key"),                                  # O(1) lookup
    BloomFilter(expected_items=1000, error_rate=0.01),  # O(1) membership
    ExpiryTTL(ttl_seconds=300, time_field="ts"),        # auto-expiry
    BoundedSize(1000),                                 # size cap
    key="ts",
)

Let the analysis choose

from fluxds import recommend
recommend({"n": 100_000, "insert": 100_000, "rank_query": 1_000_000})
# ranks skiplist / fenwick / hash_index ... and refuses a heap (no rank_query)

Your own structure, adopted as family

from fluxds import Backend, register_backend, FluxModel
from collections import OrderedDict
import itertools

@register_backend("lru", costs={"insert": "O(1)", "lookup_key": "O(1)", "delete_key": "O(1)"})
class LRUBackend(Backend):
    caps = frozenset({"extract", "peek", "remove"})
    def __init__(self):
        self.d = OrderedDict(); self._uid = itertools.count()
    def insert(self, key, item):
        uid = next(self._uid); self.d[uid] = item; return uid
    def pop_best(self):        return self.d.popitem(last=False)[1] if self.d else None
    def pop_worst(self):       return self.pop_best()
    def peek_best(self):       return next(iter(self.d.values())) if self.d else None
    def remove(self, h):       return self.d.pop(h, None) is not None
    def __len__(self):         return len(self.d)

class CacheEntry(FluxModel):
    key: str
    value: int
    class flux:
        index = ("key",); bounded = 1000; structure = "lru"

More runnable examples in examples/: showcase.py, demo.py, compose_demo.py, lld_custom_demo.py.

Run the tests

python -m unittest discover -s tests     # 27 tests, all green
# or, with pytest installed:
pytest

Where it fits (system design & ML)

• LLD / system design — LRU cache, leaderboard, rate limiter, job scheduler, delayed-payment queue, hit counter. See docs/LLD_ANALYSIS.md.
• PyTorch / ML loops — top-k checkpoint keeper, curriculum / hard-example pools (strategy switch is an O(1) swap), prioritized replay buffers. fluxds lives in the Python layer around the training loop, never in the GPU math.

Roadmap

Planned next (see docs/ROADMAP_USAGE.md, which is written usage-first — the dream API is the spec):

• flux(rows, by=..., index=...) — a no-class front door that works on plain dicts.
• Compose as keywords: unique=, ttl=, limit= directly on flux().
• flux.cache(limit, ttl, evict="lru") preset.
• search(start, goal, expand, strategy="bfs"|"dijkstra"|"astar") — one loop, many algorithms.
• DB source/sink adapters.

License

MIT © 2026 Chukkala Nikhilesh Krishna