🎴 cheat sheet

the whole framework on one page. print it out.

the three buckets

pipes

# every ZefOp works eager or lazy
trim('abcd', 'a')              ≡  'abcd' | trim('a') | collect

# data-flow arg always goes first
op(args, x)                  ≡  x | op(args)

# pipelines are values
my_chain = map(add(1)) | filter(Z > 3)
[1, 2, 3] | my_chain | collect

core ZefOps

UID prefix cheat

entities

ET.Person(name='Alice')             # single field
ET.Person(likes_={'🍔', '🍺'})       # multi (set)
ET.Person(visited_=[c1, c2])        # multi (ordered list)
ET.Person('🍃-abc...', name='Bob')  # update/create by UID
ET.Person(3, name='Bob')            # update by local index

update grammar

name='Bob'               # replace single value
likes_={'a', 'b'}        # replace set
likes_= this + {'c'}    # append
likes_= this - {'a'}    # remove
likes_={}                # clear set

types are sets

42 | is_a(Int)           # True
Int | String             # union
Int & (Z > 18)           # intersection / refinement
~String                  # complement
Array[Int | Float]       # container refinement
ET.Person['age': Int & (Z > 18), ...]   # entity refinement

FX rhythm

eff = FX.Whatever(field=1)    # build (nothing happens)
eff | run                    # fire it
eff.field                    # inspect

# partial build + insert_into
pub = FX.Publish(target=topic)
'hi' | insert_into(pub, 'content') | run

actor recipe

topic = ET.Topic('🍃-...20hex...')

@zef_function
def h(msg, state):
    return [[/* effects */], new_state]

actor = FX.StartActor(input=topic, initial_state=0, handler=h) | run
FX.Publish(target=topic, content='hi') | run
FX.QueryActorState(actor=actor) | run
FX.StopActor(actor=actor) | run

server recipe

FX.StartHTTPServer(
    routes={
      '/':                          'hi',
      '/api':                       ET.JSON(content={'ok': True}),
      ('POST', '/x'):              my_handler,
      '/ws':                        ET.WebSocket(clients_subscribe_to=[t], clients_push_into_=[t]),
    },
    port=8000,
) | run

db recipe

db = FX.CreateDB(type=DictDatabase, persistence='vault') | run
FX.UpdateDB(db=db, insert={'k': v}) | run
v = FX.QueryDB(db=db, get='k') | run

# shorthand
db['k'] = v
v = db['k']

signal recipe

s = FX.CreateSignal(value=0, history=100) | run
FX.UpdateSignal(signal=s, transform=add(1)) | run
FX.ReadSignal(signal=s) | run
FX.ReadSignalHistory(signal=s) | run

timer recipe

t = FX.StartTimer(interval=1, target=topic, content='tick') | run
# or interval='hourly' / 'daily' / ...
FX.StopTimer(timer=t) | run

parsing recipe

V = z.Forward
V = Int | String | Sequence['[', [V], ']']

input_str | parse2(V) | collect

ten commandments

1. don't forget | collect or | run
2. data-flow arg always goes first
3. UIDs must be 🍃 + 20 hex
4. field replaces, field_ replaces set, field_=this + ... appends
5. topics are mandatory for actors
6. handlers return [effects_list, new_state] exactly
7. UpdateSignal transforms must be PURE
8. don't use collect inside FX.SubscribeFX(op=...)
9. zef strings/ints from signals aren't python strings/ints (no .upper(), isinstance(x, int))
10. when in doubt: jq zefops.json

useful commands outside the framework

# find a ZefOp by name / fuzzy
jq '.[] | select(.name == "parse_json")' zefops.json
jq -r '.[].name' zefops.json | grep -iE 'json|field'

# get blurb + signature + example for anything matching
jq '.[] | select(.name | test("json"; "i")) | {name, blurb, signature_, minimal_code_examples}' zefops.json

where to look next

• notes/Overview.md — index of ALL notes
• notes/Using Zef from Python - A Practical Guide.md — canonical reference
• notes/Actors - Hands-On Guide.md — deeper actor patterns
• notes/HTTP Servers.md — full HTTP docs
• notes/Zef Effects System (Zef FX).md — every FX
• zefops.json — machine-readable catalog

if this zine helped you, write your own. the best way to learn zef is to
explain it to someone else.

🌿   happy piping!   🌿