fluree¶

Fluree: a graph database with time travel, history, and fine-grained policy.

Open a database with connect(), create or open a ledger, then write JSON-LD or Turtle and query it with SPARQL or JSON-LD:

import fluree

with fluree.connect(":memory:") as conn:
    people = conn.create("people")
    people.insert({
        "@context": {"ex": "http://example.org/"},
        "@id": "ex:alice", "ex:name": "Alice", "ex:age": 42,
    })
    for name, age in people.query(
        "PREFIX ex: <http://example.org/> "
        "SELECT ?name ?age WHERE { ?s ex:name ?name ; ex:age ?age }"
    ):
        print(name, age)

Submodules¶

Package Contents¶

class fluree.Connection(native: Connection)¶

A connection to a Fluree database. Create one with connect().

close() → None¶

Close the connection. Closing the last open connection to a directory in this process flushes pending writes and releases the directory, so another process can open it.

Afterwards the connection, and every ledger, snapshot, transaction and stream opened through it, raises InvalidRequestError; an open transaction can still be rolled back. Closing again does nothing.

create(ledger: str, *, source: str | os.PathLike[str] | None = None) → Ledger¶

Create a ledger, optionally bulk-loading RDF files into it.

source is a file or a directory of Turtle, TriG, N-Triples, N-Quads, or JSON-LD files.

ledger(ledger: str) → Ledger¶: An existing ledger. Raises NotFoundError if there is none.

ledgers() → list[str]¶: The ids of every ledger, as name:branch.

query(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Any¶

Run a query whose FROM names the ledgers to read.

SPARQL FROM <ledger> (and FROM NAMED) or a JSON-LD "from" picks the ledgers, so one query can span several; a ledger address can carry a time (<people@t:5>), and SPARQL FROM ... TO ... reads a range of history. Results, parameters, max_fuel and timeout are as for Snapshot.query(). Cypher has no FROM; run it on a ledger.

select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Result¶: A SPARQL SELECT whose FROM names the ledgers, as for query(), returning its Result; see Snapshot.select().

profile(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → QueryProfile¶: Run query() and report the fuel and time it took.

Register Iceberg tables as graph source name, mapped to RDF by mapping (R2RML, as Turtle text or a path to a Turtle file).

Give exactly one place the tables are found. Read one table directly from table_location — an s3:// prefix, or a local file:// path (local paths must lie under a directory named in the FLUREE_ICEBERG_LOCAL_ROOTS environment variable, set before the process first reads a table). Or find them through a catalog, where the mapping’s table names (or table) name the tables: a REST catalog at catalog_uri, with warehouse and auth (Bearer or OAuth2); an AWS Glue Data Catalog (Glue); or an AWS S3 Tables table bucket (S3Tables). A REST catalog’s vended credentials are used unless vended_credentials is false; otherwise, and always for Glue and S3 Tables, S3 access uses the usual AWS environment and the s3_* settings. warehouse, auth and asking for vended credentials apply to a REST catalog only, and are refused otherwise.

order_by names the column that orders a key’s rows, latest winning, for GraphSource.materialize(). model names a ledger whose policies and class hierarchy govern the source.

Register Delta Lake tables as graph source name, mapped to RDF by mapping (R2RML, as Turtle text or a path to a Turtle file).

Each mapped rr:tableName is found beneath root (a.b at <root>/a/b), at its location in tables, or through a Unity catalog. Storage credentials come from the usual AWS or Azure environment, the s3_* settings, or azure; local paths must lie under a directory named in FLUREE_ICEBERG_LOCAL_ROOTS. model and default_allow are as for map_iceberg().

map_sql(name: str, endpoint: str, mapping: str | os.PathLike[str], *, dialect: SqlDialect = 'trino', protocol: Literal['trino', 'presto'] = 'trino', catalog: str | None = None, schema: str | None = None, user: str | None = None, auth: Auth | None = None, session: collections.abc.Mapping[str, str] | None = None, allow_duplicate_subjects: bool = False, branch: str | None = None, model: str | None = None, default_allow: bool | None = None) → GraphSource¶

Register tables behind a SQL engine that speaks the Trino HTTP protocol (Trino, Presto, or a bridge to another database) as graph source name, mapped to RDF by mapping. Queries are pushed down to endpoint as SQL in dialect.

A mapped table whose subject key is not unique is reported with a warning and refused at query time, unless allow_duplicate_subjects. model and default_allow are as for map_iceberg().

graph_sources() → list[GraphSource]¶: Every graph source, by id.

graph_source(name: str) → GraphSource¶: Graph source name ("name" or "name:branch"). Raises NotFoundError if there is none.

restore(path: str | os.PathLike[str], ledger: str) → Ledger¶: Create ledger from a .flpack archive (see Ledger.archive()).

drop(ledger: str) → None¶

Delete a ledger, every branch and all history ("people"), or one branch ("people:dev").

A ledger’s first branch cannot be dropped on its own; drop the ledger. A branch that other branches were created from is hidden at once but its storage is kept until they are dropped too.

class fluree.Ledger(connection: Connection, ledger_id: str, policy: dict[str, Any] | None = None)¶

A ledger: write to it, and query its latest state or any past one.

property id: str¶: The ledger’s id, name:branch.

This ledger, governed: every read and write through the returned handle is subject to policy.

identity: the IRI (often a DID) to evaluate policy as; the policies attached to it in the ledger apply.
policy_class: apply the ledger’s policies of these classes.
policy: inline JSON-LD policy documents to apply.
values: bindings for variables the policies reference, e.g. {"?$dept": "eng"}.
default_allow: whether data no policy covers is visible and writable. None defers to the ledger’s configured default, which is to deny.

Reads filter out what policy does not allow; a write that policy does not allow raises PermissionDeniedError. So does what policy cannot filter: the commit log, info() and graphs(), and maintenance, branch and export operations.

insert(data: Data, *, format: Format | None = None, message: str | None = None) → Commit¶

Add data: JSON-LD (a dict, list, or JSON text), Turtle, or TriG.

A path is read as a file, its format taken from the extension unless format is given. message is recorded with the commit, and log() shows it. Data with nothing to add, such as [], is refused.

upsert(data: Data, *, format: Format | None = None, message: str | None = None) → Commit¶: Add data, replacing existing values of the properties it sets.

Add one node per row of rows: a pandas or polars DataFrame, or an iterable of dicts. Each column becomes a property of its row’s node; a missing value (None, NaN, NaT) adds nothing.

id: the node’s IRI, as a template over the row’s columns — "ex:person/{id}" — whose values are percent-escaped. Without one each row is a new blank node.
type: the class (or classes) every node gets.
vocab: an IRI prefix for the column names; otherwise they are read against the ledger’s default context, like JSON-LD keys.
columns: rename a column’s property (full or compact IRI), or drop the column with None.
refs: properties whose value is a reference to another node, each an IRI template — {"manager": "ex:person/{manager_id}"}.

Values convert as in insert(), keeping the DataFrame’s types: a pandas integer column with a gap is a float column, unless it has the nullable Int64 dtype. All rows commit together.

upsert_rows(rows: Any, *, id: str | None = None, type: str | list[str] | None = None, vocab: str | None = None, columns: collections.abc.Mapping[str, str | None] | None = None, refs: collections.abc.Mapping[str, str] | None = None, message: str | None = None) → Commit¶: insert_rows(), replacing existing values of the properties the rows set — to load a table again after it changed.

Apply a SPARQL UPDATE, a Cypher write, or a JSON-LD where/delete/insert.

The language is told from the text (language overrides it); a path is read as a file, .rq/.ru as SPARQL and .cypher/.cyp/.cql as Cypher. A Cypher write may be a ; script, committed all or nothing — and mixes with the other languages inside a transaction(); its $name parameters come from parameters and keyword arguments, and the records a RETURN produces are the commit’s result.

parameters and keyword arguments bind values by name: Cypher $name, and in SPARQL the variable ?name (or $name) wherever it appears, as if the value were written in its place; see Snapshot.query().

sync(data: Data, *, graph: str | None = None, format: Format | None = None, allow_empty: bool = False, dry_run: bool = False, message: str | None = None) → Commit¶

Make a graph hold exactly data, committing only the difference: facts missing from data are retracted, new ones asserted, and unchanged ones left alone. History keeps every earlier state.

graph is a named graph’s IRI; by default the default graph is synced. data is as for insert(), and Turtle, N-Triples or TriG text.

Empty data would clear the graph, so it is refused unless allow_empty is set. With dry_run nothing is committed and the returned Commit (id None) counts what would change. When data already matches, no commit is written either.

transact(fn: collections.abc.Callable[..., _T], /, *args: Any, **kwargs: Any) → _T¶

Run fn(txn, *args, **kwargs) in a Transaction and commit it, running fn again in a fresh transaction if another commit lands first. Returns what fn returns.

The way to read and then write safely: values fn reads from txn are re-read on every attempt, so what it writes never rests on data another writer has since changed. fn may commit txn itself (to pass a message); otherwise it is committed when fn returns. An exception from fn rolls the transaction back and propagates.

transaction(*, message: str | None = None) → Transaction¶

Open a Transaction: several writes, each seeing the ones before it, committed together as one commit, with message.

Use it as a context manager to commit on a clean exit and roll back on an exception:

with ledger.transaction(message="onboard bob") as txn:
    txn.insert(...)
    txn.update(...)
txn.committed  # the Commit

query(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Any¶: Query the latest state. See Snapshot.query().

select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Result¶: Run a SELECT on the latest state; see Snapshot.select().

profile(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → QueryProfile¶: Run query() and report the fuel and time it took.

stream(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, max_fuel: float | None = None, timeout: float | None = None, batch_size: int = 1000, **kwparameters: Any) → RowStream¶: Run a SELECT and read its rows as they are produced; see RowStream. max_fuel and timeout cover the whole stream.

explain(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, **kwparameters: Any) → dict[str, Any]¶: The plan the engine would run query with, without running it.

snapshot() → Snapshot¶: The latest state, frozen: every query on it sees the same data.

at(t: int | None = None, *, time: datetime.datetime | str | None = None, commit: str | None = None) → Snapshot¶

The ledger as it was at transaction t, a wall-clock time, or a commit. Give exactly one.

commit is a full commit id (Commit.id), or its hex Commit.digest or a unique prefix of it (Commit.short_id).

history(subject: str, predicate: str | None = None, *, from_t: int = 1, to_t: int | None = None) → list[Change]¶

Every change to subject between transactions from_t and to_t (default: the latest), oldest first; within one transaction, retractions come before assertions.

subject and predicate are full IRIs. Give predicate to follow one property.

log(limit: int | None = None) → list[Commit]¶: The ledger’s commits, newest first; at most limit of them.

changes(commit: CommitRef) → list[Change]¶

The facts one commit asserted and retracted.

commit is its t, its id, a prefix of its hex digest (Commit.short_id), or a Commit. Under with_policy(), facts the policy hides are left out.

branch(name: str) → Ledger¶

Create branch name from this ledger’s latest state and return it. Branch from a past state with ledger.at(...).branch(name).

The new branch shares this one’s history up to that point; from then on each changes independently, until merge() brings one’s commits into the other.

branches() → list[Branch]¶: Every branch of this ledger, by name.

merge(source: str | Ledger, *, strategy: MergeStrategy = 'take-both') → MergeResult¶

Bring the commits of branch source (a name like "dev", or its Ledger) into this branch.

If this branch has no commits of its own since source was created from it, it moves forward to source’s latest commit. Otherwise one merge commit applies source’s changes, and strategy settles properties both branches changed:

"take-both": keep both values.
"abort": change nothing and raise ConflictError.
"take-source": source’s value wins.
"take-branch": this branch’s value wins.

Preview a merge with merge_preview().

merge_preview(source: str | Ledger, *, strategy: MergeStrategy = 'take-both', details: bool = False, changes: bool = False, changes_after: str | None = None, validate: bool = True, conflicts: bool = True, max_commits: int | None = 500, max_conflicts: int | None = 200, max_changes: int | None = 500) → MergePreview¶

What merge() of source would do, without doing it.

strategy: the one to judge mergeable by.
details: include what each side wrote to each conflict.
changes: include the net facts the merge would bring in, at most max_changes of them per call; pass the result’s changes_after back to read the next page.
validate=False skips checking the merged state against the ledger’s SHACL shapes.
conflicts=False skips finding conflicts, the costly part on branches that have drifted far apart.
max_commits and max_conflicts cap the lists returned, not the work; None lifts a cap.

rebase(*, strategy: RebaseStrategy = 'take-both') → RebaseResult¶

Replay this branch’s own commits on top of the latest commit of the branch it was created from, as if it had been created from there.

strategy settles properties both branches changed: "take-both" keeps both values, "abort" changes nothing and raises ConflictError, "take-source" lets the source branch’s value win, "take-branch" lets this branch’s win, and "skip" drops each of this branch’s commits that conflicts.

revert(commits: CommitRef | list[CommitRef], *, strategy: RevertStrategy = 'abort') → RevertResult¶

Undo one commit or several, in one new commit.

A commit is named by its t, its id, a prefix of its hex digest, or a Commit. strategy settles properties that later commits changed again: "abort" changes nothing and raises ConflictError, "take-source" undoes them anyway, and "take-branch" keeps the later values. Merge commits cannot be reverted.

Preview a revert with revert_preview().

revert_preview(commits: CommitRef | list[CommitRef], *, strategy: RevertStrategy = 'abort', validate: bool = True, conflicts: bool = True, max_commits: int | None = 500, max_conflicts: int | None = 200) → RevertPreview¶: What revert() would do, without doing it. strategy is the one to judge revertable by; the other options are as for merge_preview().

Check the data against SHACL shapes, without changing anything.

By default the shapes are the ledger’s own — the ones its writes are checked against. Give shapes (JSON-LD, Turtle, or a path, as for insert()) to check against other shapes, or shapes_graph, the IRI of a named graph that holds them; include_attached adds the ledger’s own shapes to either. graph validates a named graph rather than the default graph. max_fuel and timeout bound the work, as for query().

index_status() → IndexStatus¶: How far indexing has caught up with the ledger’s commits.

index(*, timeout: float | None = None) → int¶: Index everything committed so far, waiting up to timeout seconds; returns the indexed t. Queries do not need this — they read unindexed commits too — but it makes them faster sooner. Needs a connection with background indexing.

reindex() → int¶: Rebuild the index from scratch from the commit history; returns the indexed t.

verify(*, max_commits: int | None = None) → VerifyReport¶: Check that every commit, back to the first (or the last max_commits), and the index root are present and readable.

sweep(*, dry_run: bool = False) → SweepResult¶: Delete index files that no index of any branch of this ledger references any more — left behind as indexing replaces old index files. dry_run counts them without deleting.

Export the ledger’s current data as RDF.

Writes to path and returns None, or returns the text when no path is given. format is "turtle", "trig", "ntriples", "nquads", or "jsonld"; by default it follows the path’s extension, else Turtle. The default graph is exported unless graph names one or all_graphs asks for every named graph too (use TriG or N-Quads for that). context supplies prefixes for Turtle and JSON-LD. Export the past with ledger.at(...).export(...).

archive(path: str | os.PathLike[str], *, include_indexes: bool = True) → None¶: Write the whole ledger, every commit, to a .flpack archive that Connection.restore() loads back. Without indexes the archive is smaller but the restored ledger must reindex before it is fast.

property context: dict[str, Any] | None¶

The ledger’s default JSON-LD context, or None.

Queries that omit @context (JSON-LD) or PREFIX (SPARQL) resolve their prefixes against it. Replace it with set_context().

set_context(context: dict[str, Any]) → None¶: Replace the ledger’s default JSON-LD context: a {prefix: IRI} map.

info() → dict[str, Any]¶: Ledger metadata and statistics.

graphs() → list[IRI]¶

The IRIs of the ledger’s named graphs, in the order they were first written to. The default graph and Fluree’s own graphs (commit metadata, configuration) are left out.

A graph is created by writing to it — TriG, SPARQL GRAPH, or a JSON-LD node’s "@graph" — and queried with SPARQL GRAPH or JSON-LD ["graph", iri, pattern].

drop_graph(graph: str) → Commit¶: Retract everything in named graph graph (its full IRI) in one commit. History keeps it, and the graph can be written to again. The default graph and Fluree’s own graphs cannot be dropped.

set_full_text(properties: collections.abc.Iterable[str], *, language: str = 'en', reindex: bool = True) → None¶

Make the plain-string values of properties searchable with fulltext(), analyzed in language (a BCP-47 tag; values tagged with their own language use it instead).

properties are IRIs, full or compact against the ledger’s default context; they replace the ones configured before, and an empty list turns the configuration off. The data itself is not changed.

reindex rebuilds the index so values already in the ledger are searchable. From then on, a property is searchable as values commit once an index build has seen values of it: a property configured before the ledger holds any of its values is searchable after the next index build (background indexing, or reindex()). Pass reindex=False to defer the rebuild, for instance before a bulk load, and call reindex() when ready.

full_text() → FullText | None¶: The ledger’s full-text configuration, or None without one.

class fluree.QueryProfile¶

A query’s result with what it cost: fuel (the engine’s unit of work, the unit max_fuel limits) and wall-clock time.

result: Any¶

fuel: float | None¶

time: datetime.timedelta | None¶

class fluree.Snapshot(native: Snapshot, ledger: Ledger)¶

A ledger frozen at one point in time.

Get one from Ledger.snapshot() or Ledger.at().

export(path: str | os.PathLike[str] | None = None, *, format: ExportFormat | None = None, graph: str | None = None, all_graphs: bool = False, context: dict[str, Any] | None = None) → str | None¶: Export the data as of this snapshot; see Ledger.export().

branch(name: str) → Ledger¶: Create branch name from this past state; see Ledger.branch().

property ledger: str¶: The id of the ledger this is a snapshot of, name:branch.

property t: int¶: The transaction this snapshot reflects.

Run a SPARQL, Cypher, or JSON-LD query.

The language is told from the query — a dict or JSON text is JSON-LD, text that opens like SPARQL (PREFIX, SELECT, …) is SPARQL, and text that opens like Cypher (MATCH, UNWIND, …) is Cypher — or set with language.

SPARQL SELECT and every Cypher query return a Result; Cypher nodes, relationships and paths come back as Node, Relationship and Path.
SPARQL ASK returns a bool; CONSTRUCT/DESCRIBE the constructed graph as a JSON-LD document.
A JSON-LD query returns its JSON result as Python objects.

parameters and keyword arguments bind values by name. In Cypher they are the $name parameters. In SPARQL each names a variable — ?name or $name, the same variable — and the value stands in for it wherever it appears, as if written there: query("SELECT ?s WHERE { ?s ex:name $name }", name="Alice"). A value is an IRI, BlankNode, LangString or Literal, or a Python str, int, float, bool, Decimal, datetime, date or time; a Node stands for its element_id. A SPARQL parameter the query never mentions is an error, since its misspelt variable would otherwise match anything; a JSON-LD query takes no parameters.

A Cypher write is refused here; run it with Ledger.update().

max_fuel caps the work the query may do (see profile() for what a query costs); past it the query stops with ResourceLimitError. timeout is in seconds; past it the query is cancelled with QueryTimeoutError.

select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Result¶: Run a SPARQL SELECT or a Cypher query and return its Result — query() for the queries whose result is a table, typed as one. Anything else (ASK, CONSTRUCT, JSON-LD) raises InvalidRequestError without running. Parameters, max_fuel and timeout are as for query().

profile(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → QueryProfile¶: Run query() and report the fuel and time it took.

stream(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, max_fuel: float | None = None, timeout: float | None = None, batch_size: int = 1000, **kwparameters: Any) → RowStream¶: Run a SELECT and read its rows as they are produced; see RowStream.

explain(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, **kwparameters: Any) → dict[str, Any]¶: The plan the engine would run query with, without running it.

class fluree.Transaction(ledger: Ledger, native: Transaction, message: str | None)¶

Writes staged one at a time and committed together as one commit.

Get one from Ledger.transaction(). Each write applies over the ones before it — an update’s WHERE sees an earlier insert — and is checked as it is staged, so a write that is invalid or that policy does not allow raises at once and is left out; the transaction carries on without it. Queries on the transaction read the staged state; nothing is visible on the ledger until commit().

A fact that one write adds and a later one removes (or the reverse) is left out of the commit, so the commit holds only the net change.

The writes stage against the ledger as it was when the transaction began. If another commit lands first:

a transaction that was only written to is staged again on top of it — each update’s WHERE matches the new data;
a transaction that was also read (query, explain, or the RETURN records of a Cypher write) raises ConflictError on commit(), since what was read may have decided what was written. Run it again, or use Ledger.transact(), which does.

As a context manager it commits on a clean exit and rolls back on an exception; committed holds the resulting Commit.

committed: Commit | None = None¶

insert(data: Data, *, format: Format | None = None) → None¶: Stage an insert; see Ledger.insert().

upsert(data: Data, *, format: Format | None = None) → None¶: Stage an upsert; see Ledger.upsert().

insert_rows(rows: Any, *, id: str | None = None, type: str | list[str] | None = None, vocab: str | None = None, columns: collections.abc.Mapping[str, str | None] | None = None, refs: collections.abc.Mapping[str, str] | None = None) → None¶: Stage Ledger.insert_rows().

upsert_rows(rows: Any, *, id: str | None = None, type: str | list[str] | None = None, vocab: str | None = None, columns: collections.abc.Mapping[str, str | None] | None = None, refs: collections.abc.Mapping[str, str] | None = None) → None¶: Stage Ledger.upsert_rows().

update(transaction: str | dict[str, Any] | os.PathLike[str], parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, **kwparameters: Any) → Result | None¶: Stage an update — SPARQL UPDATE, a Cypher write, or JSON-LD; see Ledger.update(). A Cypher write returns the records of its RETURN (None without one); receiving them counts as reading the transaction.

query(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Any¶: Query the staged state; see Snapshot.query().

select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Result¶: Run a SELECT on the staged state; see Snapshot.select().

explain(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, **kwparameters: Any) → dict[str, Any]¶: The plan query would run with over the staged state.

commit(*, message: str | None = None) → Commit¶: Commit the staged writes as one commit, recording message (by default the one the transaction was opened with). The transaction is closed afterwards, even if the commit fails.

rollback() → None¶: Discard the staged writes and close the transaction.

fluree.connect(path: str | os.PathLike[str] | None = None, *, config: dict[str, Any] | None = None, indexing: bool = True) → Connection¶

Open a Fluree database, creating it if needed.

Give a directory path, or ":memory:" for a database that lives only as long as the connection. indexing=False turns off background indexing in this process.

Connections in one process to the same directory share one engine, so threads may each open their own; they must agree on indexing. While any of them is open, another process that opens the directory raises DatabaseInUseError: run fluree server on it to share it across processes. A config connection to file storage is not shared: while it is open, opening that storage again raises DatabaseInUseError, in this process too.

For anything else — S3 or split commit/index storage, a DynamoDB nameservice, encryption at rest, values read from environment variables — pass a JSON-LD connection config instead of a path; it is the same document fluree server takes with --connection-config.

The connection is a context manager that closes it on exit:

with fluree.connect("./data") as conn:
    ledger = conn.create("people")

class fluree.Node(element_id: IRI | BlankNode, labels: frozenset[str], properties: dict[str, Any])¶

Bases: collections.abc.Mapping[str, Any]

A node: its properties (read like a dict), its labels — the node’s classes — and element_id, its identity: the same IRI (or BlankNode) a SPARQL query returns for it.

element_id¶

labels¶

class fluree.Path(nodes: tuple[Node, ...], relationships: tuple[Relationship, ...])¶

A walk through the graph: nodes and the relationships between them, in order. Iterating a path yields its relationships.

nodes¶

relationships¶

property start_node: Node¶: The node the path starts at.

property end_node: Node¶: The node the path ends at.

class fluree.Relationship(element_id: IRI | BlankNode | None, type: str, start_node: Node, end_node: Node, properties: dict[str, Any])¶

Bases: collections.abc.Mapping[str, Any]

A relationship: its properties (read like a dict), its type, and the nodes it runs from and to. element_id is its identity when it has one (every relationship a Cypher CREATE makes does).

start_node and end_node carry their properties and labels when the same result returned those nodes; otherwise only their element_id.

element_id¶

type¶

start_node¶

end_node¶

property nodes: tuple[Node, Node]¶: (start_node, end_node).

class fluree.Branch¶

A branch of a ledger. Open it with conn.ledger(branch.id).

source is the branch it was created from, None for the ledger’s first branch. t and head are its latest commit’s t and id.

name: str¶

id: str¶

source: str | None¶

t: int¶

head: str | None¶

class fluree.Change¶

One fact asserted or retracted at transaction t: subject’s predicate gained (op == "assert") or lost (op == "retract") value, in the default graph or the named graph.

t is None for the net changes a merge preview reports, which span several commits.

t: int | None¶

op: Literal['assert', 'retract']¶

subject: IRI | BlankNode¶

predicate: IRI¶

value: Any¶

graph: IRI | None = None¶

class fluree.Commit¶

The outcome of a transaction.

id is the commit’s content id (a CID). digest is its hex hash, the form fluree log shows: like a git SHA, a unique prefix of it (short_id, say) identifies the commit. A prefix of id does not, since every CID starts with the same header characters.

id and digest are None when the transaction changed nothing, in which case no commit was written and t is the ledger’s unchanged t.

time and message are filled in for commits read from Ledger.log() and the previews. result holds the records a Cypher write’s RETURN produced.

t: int¶

id: str | None¶

digest: str | None¶

asserts: int¶

retracts: int¶

time: datetime.datetime | None = None¶

message: str | None = None¶

result: Result | None = None¶

property short_id: str | None¶: The first 12 hex digits of digest, as fluree log prints them.

class fluree.Conflict¶

A property of a subject that both sides changed.

source and target are what each side wrote to it, when the preview was asked for details; otherwise None.

subject: IRI | BlankNode¶

predicate: IRI¶

graph: IRI | None¶

source: list[Change] | None = None¶

target: list[Change] | None = None¶

class fluree.FullText¶

A ledger’s full-text configuration (see Ledger.set_full_text()): the properties whose plain-string values fulltext() scores, and the language untagged values are analyzed in.

properties: tuple[IRI, ...]¶

language: str | None¶

class fluree.IndexStatus¶

Where indexing of a ledger stands. Commits after index_t (up to commit_t) are queryable but not yet indexed. phase is "idle", "pending" or "in_progress"; enabled is false on a connection without background indexing (":memory:", or indexing=False).

index_t: int¶

commit_t: int¶

enabled: bool¶

phase: str¶

error: str | None¶

class fluree.MergePreview¶

What merging source into target would do, without doing it.

ahead: commits on source that target lacks, newest first; behind: the reverse. Each list may be capped; ahead_count and behind_count are the full counts.
ancestor_t: the t of the last commit the two share.
fast_forward: target has no commits of its own since, so the merge just moves it forward.
conflicts: properties both sides changed (may be capped; conflict_count is the full count).
mergeable: the merge would succeed under the previewed strategy, including the ledger’s SHACL shapes; when it would fail validation, violations holds the report.
changes: the net facts the merge would bring in, when asked for. Pass changes_after back as changes_after= to read the next page.

source: str¶

target: str¶

fast_forward: bool¶

mergeable: bool¶

ancestor_t: int | None¶

ahead: list[Commit]¶

ahead_count: int¶

behind: list[Commit]¶

behind_count: int¶

conflicts: list[Conflict]¶

conflict_count: int¶

violations: str | None¶

changes: list[Change] | None¶

changes_after: str | None¶

class fluree.MergeResult¶

The outcome of Ledger.merge().

A fast_forward merge moved target to source’s latest commit without writing one; otherwise id is the new merge commit. t is target’s t afterwards. conflicts counts the properties strategy resolved.

source: str¶

target: str¶

fast_forward: bool¶

t: int¶

id: str¶

digest: str¶

conflicts: int¶

strategy: str | None¶

class fluree.RebaseResult¶

The outcome of Ledger.rebase().

Of the branch’s total own commits, replayed were rewritten on top of its source’s latest commit (source_t) and skipped were dropped by the "skip" strategy. conflicts lists (t, properties, resolution) for each commit that conflicted, t being the commit’s original t; failures lists (t, error) for commits that failed validation once replayed. A fast_forward rebase had no commits of its own to replay.

fast_forward: bool¶

replayed: int¶

skipped: int¶

total: int¶

source_t: int¶

conflicts: list[tuple[int, int, str]]¶

failures: list[tuple[int, str]]¶

class fluree.RevertPreview¶

What reverting would do, without doing it.

commits are the commits that would be undone, newest first (may be capped; commit_count is the full count). conflicts are properties changed again since those commits. revertable says whether the revert would succeed under the previewed strategy and the ledger’s SHACL shapes; when validation would fail, violations holds the report.

revertable: bool¶

commits: list[Commit]¶

commit_count: int¶

conflicts: list[Conflict]¶

conflict_count: int¶

violations: str | None¶

class fluree.RevertResult¶

The outcome of Ledger.revert().

reverted holds the ids of the undone commits, newest first. When committed is false the commits had nothing left to undo, no commit was written, and t/id describe the unchanged head. conflicts counts the properties changed since that strategy resolved.

committed: bool¶

t: int¶

id: str¶

digest: str¶

reverted: list[str]¶

conflicts: int¶

strategy: str¶

class fluree.SweepResult¶

The outcome of Ledger.sweep(): orphans index files no longer referenced, of which reclaimed were deleted (none on a dry run); failures lists (file, error) for those that could not be.

dry_run: bool¶

orphans: int¶

reclaimed: int¶

failures: list[tuple[str, str]]¶

class fluree.ValidationReport¶

The outcome of Ledger.validate(). conforms is false when any result is a violation. shape_count is how many shapes were checked — 0 means no shapes were found, so the data trivially conforms. t is the state that was validated.

conforms: bool¶

results: list[ValidationResult]¶

shape_count: int¶

t: int¶

class fluree.ValidationResult¶

One way the data fails a SHACL shape.

focus is the node that failed and path the property, when the constraint is on one; value is the offending value, when there is one. severity is "violation", "warning" or "info". shape is the node shape and component the SHACL constraint component (sh:MinCountConstraintComponent, …) that produced it.

focus: Any¶

path: IRI | None¶

message: str¶

severity: str¶

value: Any¶

shape: IRI | BlankNode¶

component: IRI¶

class fluree.VerifyReport¶

The outcome of Ledger.verify().

severity is "healthy", "provenance" (a raw transaction is missing: the data is intact) or "chain" (a commit or the index root is missing or unreadable). problems describes each one, as a dict with a kind. truncated means max_commits stopped the check early.

severity: str¶

head_t: int¶

index_t: int¶

commits_checked: int¶

truncated: bool¶

problems: list[dict[str, Any]]¶

property healthy: bool¶: Whether nothing was found missing or unreadable.

fluree.set_log_level(level: str | int) → None¶

Send the engine’s log events at level and above to the fluree.engine logger: a logging level ("INFO", logging.DEBUG), "TRACE" for the most detail, or "OFF". Warnings and errors are sent by default.

This sets what the engine produces; the logger’s own level and handlers still decide what is shown. Below the level, the engine skips the events entirely.

class fluree.Record¶

Bases: tuple

One result row: a tuple in the query’s column order that can also be read by column name — record["name"] or record.name — with get, keys, values, items and data.

Attribute access gives way to the methods: a column named data or count is read as record["data"]. Unbound values are None.

get(key: str, default: Any = None) → Any¶: The value of column key (None when unbound), or default when the result has no column key.

index(key: str | int) → int¶: The position of column key.

keys() → list[str]¶: The column names, in order.

values(*keys: str | int) → list[Any]¶: The values of columns keys (names or positions), or of every column.

items(*keys: str | int) → list[tuple[str, Any]]¶: (column, value) pairs for columns keys, or for every column.

data(*keys: str | int) → dict[str, Any]¶: The record as a dict of plain values: nodes become their property dicts, relationships (start properties, type, end properties).

class fluree.Result(keys: list[str], records: list[Record])¶

Bases: collections.abc.Sequence[Record]

The records a SPARQL SELECT or a Cypher statement returned, in order. Iterate it, index it, or read it as a whole: single() for the one record, first() for the first, value() for one column, data() for dicts, to_pandas() for a DataFrame.

keys() → list[str]¶: Column names in projection order (SPARQL variables without ?).

property columns: list[str]¶: The same as keys().

single() → Record¶: The one record; InvalidRequestError unless there is exactly one. For “the first record, if any”, use first().

first() → Record | None¶: The first record, or None when there are none.

value(key: str | int = 0, default: Any = None) → list[Any]¶: One column’s values, default where a record lacks it.

values(*keys: str | int) → list[list[Any]]¶: Every record’s values, as lists; see Record.values().

data(*keys: str | int) → list[dict[str, Any]]¶: Every record as a dict of plain values; see Record.data().

to_pandas() → pandas.DataFrame¶: The records as a pandas DataFrame, one column per key. Nodes and relationships stay objects.

to_df¶

to_polars() → polars.DataFrame¶: The records as a polars DataFrame, one column per key. Nodes and relationships become object columns.

class fluree.RowStream(native: RowStream, batch_size: int)¶

Bases: collections.abc.Iterator[Record]

The records of a SELECT, read as the query produces them.

Iterate it like a Result; memory stays flat however many records the query returns. The query runs until the stream ends, close() is called, or the stream is garbage-collected — so when reading may stop early, use it as a context manager, which closes it at the end of the block:

with ledger.stream(query) as rows:
    for row in rows:
        if done(row):
            break

keys() → list[str] | None¶: Column names in projection order (for SELECT *, known once the first record is read).

property columns: list[str] | None¶: The same as keys().

close() → None¶: Stop the query; the stream yields nothing more.

class fluree.AzureServicePrincipal¶

Microsoft Entra service-principal credentials for Delta tables in ADLS Gen2 or OneLake.

tenant_id: str¶

client_id: str¶

client_secret: Secret¶

class fluree.Bearer¶

Authenticate with a static bearer token.

token: Secret¶

class fluree.EnvVar¶

A secret read from environment variable name where the source is read, so the secret itself is never stored with the graph source.

name: str¶

class fluree.Glue¶

An AWS Glue Data Catalog that Iceberg tables are found through: each mapped rr:tableName is a Glue <database>.<table>, whose current metadata file Glue names. The catalog call and the S3 reads use the usual AWS environment, never vended credentials. region is the catalog’s region (else s3_region, else the AWS region chain); catalog_id reads another account’s catalog.

region: str | None = None¶

catalog_id: str | None = None¶

class fluree.GraphSource(connection: Connection, raw: collections.abc.Mapping[str, Any])¶

A graph source: tables mapped to RDF and queried in place.

Create one with Connection.map_iceberg(), Connection.map_delta() or Connection.map_sql(), or look one up with Connection.graph_source(). Query it on its own here, or with other graph sources and ledgers through Connection.query(), naming it in FROM <name:main> (SPARQL) or "from" (JSON-LD).

id is name:branch, and kind the kind of source, such as "iceberg", "delta" or "sql".

id: str¶

name: str¶

branch: str¶

kind: str¶

query(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Any¶: Run a SPARQL or JSON-LD query of this source; see Snapshot.query(). Cypher reads ledgers only.

select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → Result¶: Run a SPARQL SELECT of this source; see Snapshot.select().

profile(query: Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) → QueryProfile¶: Run query() and report the fuel and time it took.

materialize(into: str, *, full: bool = False) → MaterializeResult¶

Copy this source’s rows into ledger into (created if missing) as ordinary ledger data, so they gain history, policy and indexes.

Each pass reads only what the source has added since the last pass into into, where the table’s history allows; full re-reads everything. A row whose subject reappears replaces it. Iceberg sources only.

drop() → None¶: Remove this graph source. The tables themselves are not touched.

class fluree.MaterializeResult¶

One GraphSource.materialize() pass.

snapshot is the source snapshot now reflected in the ledger (None for a table with no snapshots yet). incremental says whether only what the source added since the last pass was read; committed whether anything changed in the ledger.

snapshot: int | None¶

incremental: bool¶

committed: bool¶

rows_read: int¶

subjects_upserted: int¶

subjects_retracted: int¶

class fluree.OAuth2¶

Authenticate with OAuth2 client credentials, fetching (and refreshing) a token from token_url (which a Unity catalog can leave to its default).

token_url: str | None¶

client_id: str¶

client_secret: Secret¶

scope: str | None = None¶

audience: str | None = None¶

class fluree.S3Tables¶

An AWS S3 Tables table bucket that Iceberg tables are found in, named by its ARN (arn:aws:s3tables:<region>:<account>:bucket/<name>): each mapped rr:tableName is a <namespace>.<table> in the bucket. The region defaults to the ARN’s. Reads use the usual AWS environment.

table_bucket_arn: str¶

region: str | None = None¶

class fluree.Unity¶

A Databricks Unity Catalog that Delta tables are found and read through. catalog and schema complete table names of fewer than three parts. An OAuth2 token_url defaults to the workspace’s own endpoint.

uri: str¶

catalog: str | None = None¶

schema: str | None = None¶

auth: Auth | None = None¶

class fluree.IRI¶

Bases: str

An IRI. Compares equal to the same IRI as a plain str.

class fluree.BlankNode¶

Bases: str

A blank node, by its label (without the _: prefix).

class fluree.LangString¶

Bases: str

A language-tagged string: a str with a language tag.

It compares like a string: equal to a plain str with the same text, and to another LangString with the same text and the same tag (compared case-insensitively, as BCP 47 tags are).

language: str¶

class fluree.Literal¶

A literal with no lossless Python type, as lexical form and datatype.

value: str¶

datatype: str¶

class fluree.Vector¶

Bases: tuple

An embedding vector: a tuple of floats, stored as f:embeddingVector.

Build one from any sequence of numbers or a numpy array. Insert it as a property value, pass it as a query parameter, and read it back from queries; numpy.asarray(vector) gives a float32 array. Values are stored at float32 precision. A one-dimensional numpy array is taken as a vector wherever a Vector is.

exception fluree.ConflictError¶

Bases: FlureeError

A concurrent change won; retrying against the new state may succeed.

When a commit lost a race, expected_t is the ledger t it was built on and head_t the t another commit had moved it to; both are None for other conflicts.

expected_t: int | None = None¶

head_t: int | None = None¶

exception fluree.DatabaseInUseError¶

Bases: FlureeError

The database directory is open in another process.

One process opens a directory at a time, and its threads share that connection. To use the same data from several processes, run fluree server on the directory and send requests to it. pid is the other process’s id, or None when it could not be read.

pid: int | None = None¶

exception fluree.FlureeError¶

Bases: Exception

Base class for all Fluree errors.

status is the HTTP-style status code the engine assigned the error, or None when the error did not come from the engine.

status: int | None = None¶

exception fluree.InvalidRequestError¶

Bases: FlureeError, ValueError

The query, transaction, or argument is malformed or invalid.

exception fluree.NotFoundError¶

Bases: FlureeError, LookupError

A ledger, branch, commit, or graph does not exist.

exception fluree.PermissionDeniedError¶

Bases: FlureeError, PermissionError

Policy denied the operation.

exception fluree.QueryTimeoutError¶

Bases: FlureeError, TimeoutError

The operation ran past its time limit.

exception fluree.ResourceLimitError¶

Bases: FlureeError

The operation exceeded a fuel, memory, or size limit.

exception fluree.ShaclViolationError¶

Bases: InvalidRequestError

A write that would leave data breaking the ledger’s SHACL shapes.

violations lists each way it would, as the same ValidationResult objects Ledger.validate() reports, so a rejected write is handled like a failed validation.

property violations: list[ValidationResult]¶: Each way the write breaks the shapes, as ValidationResult objects.

exception fluree.UniqueConstraintError¶

Bases: InvalidRequestError

A write that would give two subjects the same value of a property the ledger requires to be unique (f:enforceUnique).

value is the value as text; graph is the named graph’s IRI, or None for the default graph.

property: IRI¶

value: str¶

graph: IRI | None¶

existing_subject: IRI¶

new_subject: IRI¶