Language Bindings

use arcflow::{open, open_concurrent, compile, Label, Properties};
 
// In-memory graph — direct store access
let mut store = arcflow::open();
let id = store.create_node(
    vec![Label::new("Person")],
    Properties::new(),
);
 
// Thread-safe concurrent access — recommended for applications
let db = arcflow::open_concurrent();
db.execute("CREATE (n:Person {name: 'Alice', age: 30})").unwrap();
 
let result = db.execute("MATCH (n:Person) RETURN n.name").unwrap();
for row in &result.rows {
    println!("{}", row.get("name").unwrap());
}

let db = arcflow::open_concurrent();
db.execute("CREATE (n:Person {name: 'Alice'})").unwrap();
 
// Pre-compile for repeated execution
let query = arcflow::compile("MATCH (n:Person) RETURN n.name").unwrap();
 
// Execute compiled query
let result = db.execute("CALL algo.pageRank()").unwrap();
assert!(result.rows.len() > 0);

from arcflow import ArcFlow
 
# In-memory or persistent
db = ArcFlow()                  # in-memory
db = ArcFlow("/tmp/mydb")       # persistent (WAL-journaled)
 
# Execute queries — typed parameters, no manual string escaping
db.execute("CREATE (n:Person {name: 'Alice', age: 30})")
result = db.execute(
    "MATCH (p:Person) WHERE p.name = $name RETURN p.age",
    params={"name": "Alice"},
)
for row in result:
    print(row)
 
# Algorithms work the same way
db.execute("CALL algo.pageRank()")
 
# Context-manager for deterministic cleanup
with ArcFlow() as db:
    db.execute("CALL db.demo()")
 
# Version — both return the same string. See /docs/reference/versioning
import arcflow
print(arcflow.__version__)        # "0.8.0"
print(ArcFlow.version())          # "0.8.0"

import pyarrow as pa
 
# Tier 1 — Python tuples (most common shape)
entity_ids = db.bulk_create_nodes([
    (["Entity"], {"id": i, "x": float(i), "y": 0.0})
    for i in range(95)
])
frame_ids = db.bulk_create_nodes([
    (["Frame"], {"id": f"frame_{t}", "ts_ms": t * 16})
    for t in range(46_000)
])
edges = [(eid, fid, {"x": float(ei), "y": float(fi)})
         for fi, fid in enumerate(frame_ids)
         for ei, eid in enumerate(entity_ids)]
db.bulk_create_relationships("OBSERVED_AT", edges)
 
# Tier 2 — Arrow-direct (no JSON marshal, saturates the engine)
tbl = pa.table({
    "id": pa.array(range(95), type=pa.int64()),
    "x":  pa.array([float(i) for i in range(95)], type=pa.float64()),
})
db.bulk_create_nodes_from_arrow("Entity", tbl)
# Spatial Struct columns ({x: f64, y: f64[, z]}) round-trip as native Point/Point3d.
 
# Tier 3 — File-direct (no Python in the loop at all)
n_loaded = db.load_parquet("data/tracking.parquet", "TrackingObs")
db.load_csv("data/players.csv", "Player", chunk_size=100_000)

stmt = db.prepare("MATCH (e:Entity {id: $eid}) RETURN e.x, e.y, e.speed")
for eid in entity_ids:
    row = next(iter(stmt.execute({"eid": eid})))
stmt.close()

result = db.execute("MATCH (p:Player) RETURN p.jersey, p.name, p.age")
tbl = result.to_arrow()      # pyarrow.RecordBatch
df  = result.to_polars()     # polars.DataFrame  (Arrow-backed, zero-copy)
df  = result.to_pandas()     # pandas.DataFrame   (Arrow-backed since pandas 2)
 
# Direct DuckDB interop:
import duckdb
con = duckdb.connect()
con.register("players", tbl)
con.execute("SELECT count(*) FROM players WHERE jersey > 50").fetchone()

# Capture the seq before running a decision loop
anchor_seq = db.execute("RETURN walSeq() AS s").one()["s"]
 
# … mutations happen …
 
# Replay the world as it looked when the decision was made
result = db.query_at(
    "MATCH (a:Account {id: 'ACC-789'}) RETURN a.balance, a.risk_score",
    seq=anchor_seq,
)

db.execute("""
    CREATE LIVE VIEW high_conf AS
    MATCH (n:Detection) WHERE n.confidence > 0.9
    RETURN n.id AS id, n.confidence AS conf
""")
 
with db.subscribe("high_conf") as sub:
    for event in sub:                  # blocks until next frontier advance
        for row in event["added"]:
            handle_new_detection(row)
        for row in event["removed"]:
            retract(row)

last_seq = 0
while True:
    changes = db.changes_since(last_seq, limit=1_000)
    if not changes:
        time.sleep(0.5)
        continue
    publish(changes)
    last_seq = max(int(c["sequence"]) for c in changes)

from arcflow import ArcFlow
from arcflow.bus import Bus, EventPublisher, START_FROM_BEGINNING
 
db = ArcFlow()
bus = Bus(db)
 
bus.create_topic("orders.us")
seq = bus.publish("orders.us", {"order_id": 42, "amount": 100.0})
 
# Durable consumer — pull-style, per-consumer offset
bus.register_consumer("billing", "orders.us", start=START_FROM_BEGINNING)
for event in bus.pending("billing", "orders.us"):
    process(event.data)
    bus.ack("billing", "orders.us", event.seq)
 
# Wildcard consumer — fans out across every matching topic
tracks = bus.register_pattern_consumer("worker", "all_tracks.*")
for topic, event in tracks.drain():
    handle(topic, event.data)
 
# Synchronous request-reply
reply = bus.request("video.query.status", {"q": "n_cams"}, timeout_ms=5_000)
print(reply.data["count"])

pub = EventPublisher(bus, topic_prefix="framing.alert")
pub.set_rate_limit("ball_lost", interval_ms=10_000)
pub.emit("ball_lost", {"conf": 0.2}, camera="cam1", severity="warning")
# -> publishes to framing.alert.warning.ball_lost.cam1 (auto-creates the topic)

from arcflow.repository import BaseRepository, RepositoryTimeoutError
 
class PlayerRepo(BaseRepository):
    def get(self, player_id: int) -> dict | None:
        return self.execute_one(
            "MATCH (p:Player {id: $id}) RETURN p",
            {"id": player_id},
        )
 
with PlayerRepo("/var/lib/arcflow") as repo:
    alice = repo.get(42)

from arcflow import ArcFlow
from arcflow.stream_fn import StreamFn, Kind
 
db = ArcFlow()
pan_delta = StreamFn(db, "pan_delta", Kind.Delta)
ts_delta  = StreamFn(db, "ts_delta",  Kind.Delta)
 
for pan, ts in samples:
    pan_out = pan_delta.feed(pan)
    ts_out  = ts_delta.feed(ts)
    if pan_out and ts_out:
        pan_rate = pan_out[0].fn_value / ts_out[0].fn_value

from concurrent.futures import ThreadPoolExecutor
 
queries = [
    "MATCH (p:Person) WHERE p.age > 30 RETURN count(*)",
    "MATCH (p:Player) RETURN avg(p.jersey)",
    # ...
]
with ThreadPoolExecutor(max_workers=8) as pool:
    results = list(pool.map(db.execute, queries))   # reads — lock-free

import { open, openInMemory, ArcflowError } from '@ozinc/arcflow'
 
// In-memory (testing, short-lived)
const db = openInMemory()
 
// Persistent (WAL-journaled, survives crashes)
const db = open('./data/graph')
 
// Queries return typed values — numbers are numbers, not strings
db.mutate("CREATE (n:Person {name: 'Alice', age: 30})")
const result = db.query("MATCH (n:Person) RETURN n.name, n.age")
result.rows[0].get('name')  // "Alice"
result.rows[0].get('age')   // 30 (number, not "30")
 
// Parameters (prevent injection)
db.query("MATCH (n:Person {name: $name}) RETURN n", { name: 'Alice' })
 
// Atomic batch mutations
db.batchMutate([
  "MERGE (a:Person {id: 'p1', name: 'Alice'})",
  "MERGE (b:Org {id: 'o1', name: 'Acme'})",
])
 
// Paginated cursor for large result sets
const cursor = db.cursor('MATCH (n:Log) RETURN n.ts ORDER BY n.ts', undefined, 500)
let page
while ((page = cursor.next()) !== null) process(page.rows)
cursor.close()
 
// Live subscriptions — callback fires with added/removed events on each relevant mutation
const sub = db.subscribe(
  'MATCH (n:Alert) WHERE n.level = "critical" RETURN n.id, n.message',
  ({ added, removed }) => console.log('new alerts', added)
)
// later:
sub.cancel()
 
// Sync
db.syncPending()    // 0 if up to date
db.fingerprint()    // hash of current state
 
// Error handling
try { db.query("INVALID") } catch (e) {
  if (e instanceof ArcflowError) console.log(e.code, e.category, e.suggestion)
}
 
db.close()

import { openInMemory } from '@ozinc/arcflow'
import { useQuery, useLiveQuery } from '@arcflow/react'
 
const db = openInMemory()
db.mutate("CREATE (n:Alert {level: 'critical', message: 'Disk 95%'})")
 
function AlertPanel() {
  // One-time query
  const { data, loading, error } = useQuery(db, 'MATCH (n:Alert) RETURN n.level, n.message')
 
  // Live subscription — re-renders on every graph mutation that affects results
  const { rows } = useLiveQuery(db, 'MATCH (n:Alert) WHERE n.level = "critical" RETURN n.message')
 
  if (loading) return <div>Loading...</div>
  return (
    <ul>
      {rows?.map((row, i) => <li key={i}>{String(row.message)}</li>)}
    </ul>
  )
}

#include "arcflow.h"
 
// Open runtime (NULL for in-memory, or path for persistent)
arcflow_runtime_t* rt = arcflow_open(NULL);
 
// Open session
arcflow_session_t* session = arcflow_session_open(rt);
 
// Execute query
arcflow_result_t* result = arcflow_execute(session, "MATCH (n) RETURN n.name");
 
// Read results
int64_t rows = arcflow_result_row_count(result);
int64_t cols = arcflow_result_column_count(result);
 
for (int64_t r = 0; r < rows; r++) {
    const char* name = arcflow_result_get_string(result, r, 0);
    printf("%s\n", name);
}
 
// Cleanup
arcflow_result_free(result);
arcflow_session_close(session);
arcflow_close(rt);
 
// Error handling
const char* err = arcflow_last_error();

#include "arcflow.hpp"
 
int main() {
    // RAII: runtime auto-freed on scope exit
    arcflow::Runtime rt;                     // in-memory
    // arcflow::Runtime rt("/path/to/data"); // persistent
 
    auto session = rt.session();
 
    session.execute("CREATE (n:Person {name: 'Alice'})");
    auto result = session.execute("MATCH (n:Person) RETURN n.name");
 
    for (int64_t r = 0; r < result.row_count(); ++r) {
        std::cout << result.get(r, 0) << std::endl;  // "Alice"
    }
    // result, session, rt auto-freed on scope exit
}

g++ -std=c++17 \
    -I include/arcflow \
    main.cpp \
    -L /path/to/arcflow/lib -larcflow \
    -o myapp