fluree.aio

The fluree API for asyncio.

The same classes and methods as fluree, with every call that reaches the engine a coroutine:

import fluree.aio

async with fluree.aio.connect("./data") as conn:
    people = await conn.create("people")
    await people.insert({...})
    rows = await people.query("SELECT ...")
    async for row in people.stream("SELECT ..."):
        ...
    async with people.transaction() as txn:
        await txn.insert({...})

Each call runs on a worker thread from the event loop’s default executor while the loop carries on. The engine releases the GIL, so calls from many tasks run in parallel.

Cancelling a task that is waiting on a query — asyncio.timeout, a client that disconnects — stops the query in the engine too. A write cannot be stopped part way: cancelling the task stops the wait, and the write still completes or fails on its own.

Module Contents

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

Open a database; see fluree.connect(). Await it, or use it with async with to close the connection at the end of the block.

class fluree.aio.Connection(sync: fluree.Connection)

A connection to a Fluree database; see fluree.Connection.

async close() → None

Flush pending writes and release the database; see fluree.Connection.close().

async exists(ledger: str) → bool

Whether ledger exists (ledger in conn for the sync API).

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

Create a ledger, optionally bulk-loading RDF files into it; see fluree.Connection.create().

async ledger(ledger: str) → Ledger

An existing ledger; see fluree.Connection.ledger().

async ledgers() → list[str]

The ids of every ledger, as name:branch; see fluree.Connection.ledgers().

async 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; see fluree.Connection.query().

async 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 fluree.Connection.select().

async 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; see fluree.Connection.profile().

async map_iceberg(name: str, mapping: str | os.PathLike[str], **options: Any) → GraphSource

Register Iceberg tables; see fluree.Connection.map_iceberg().

async map_delta(name: str, mapping: str | os.PathLike[str], **options: Any) → GraphSource

Register Delta tables; see fluree.Connection.map_delta().

async map_sql(name: str, endpoint: str, mapping: str | os.PathLike[str], **options: Any) → GraphSource

Register SQL tables; see fluree.Connection.map_sql().

async graph_sources() → list[GraphSource]

Every graph source, by id; see fluree.Connection.graph_sources().

async graph_source(name: str) → GraphSource

Graph source name ("name" or "name:branch"); see fluree.Connection.graph_source().

async restore(path: str | os.PathLike[str], ledger: str) → Ledger

Create ledger from a .flpack archive; see fluree.Connection.restore().

async drop(ledger: str) → None

Delete a ledger, every branch and all history ("people"), or one branch ("people:dev"); see fluree.Connection.drop().

class fluree.aio.Ledger(connection: Connection, sync: fluree.Ledger)

A ledger; see fluree.Ledger.

property id: str

The ledger’s id, name:branch.

with_policy(*, identity: str | None = None, policy_class: str | list[str] | None = None, policy: dict[str, Any] | list[Any] | None = None, values: dict[str, Any] | None = None, default_allow: bool | None = None) → Ledger

This ledger, governed; see fluree.Ledger.with_policy().

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

Add data; see fluree.Ledger.insert().

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

Add data, replacing existing values of the properties it sets; see fluree.Ledger.upsert().

async insert_rows(rows: Any, **options: Any) → Commit

See fluree.Ledger.insert_rows().

async upsert_rows(rows: Any, **options: Any) → Commit

See fluree.Ledger.upsert_rows().

async update(transaction: str | dict[str, Any] | os.PathLike[str], parameters: collections.abc.Mapping[str, Any] | None = None, *, language: Language | None = None, message: str | None = None, **kwparameters: Any) → Commit

Apply a SPARQL UPDATE, a Cypher write, or a JSON-LD where/delete/insert; see fluree.Ledger.update().

async 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; see fluree.Ledger.sync().

async transact(fn: collections.abc.Callable[..., collections.abc.Awaitable[T]], /, *args: Any, **kwargs: Any) → T

Run await fn(txn, *args, **kwargs) in a Transaction and commit it, running fn again if another commit lands first; see fluree.Ledger.transact().

transaction(*, message: str | None = None) → _Opening[Transaction]

Open a Transaction. Await it, or use it with async with to commit on a clean exit and roll back on an exception.

async 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 fluree.Ledger.query().

async 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 fluree.Ledger.select().

async 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; see fluree.Ledger.profile().

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

Read a SELECT’s rows as they are produced: async for row in ledger.stream(...). The query starts on the first read.

async 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; see fluree.Ledger.explain().

async snapshot() → Snapshot

The latest state, frozen; see fluree.Ledger.snapshot().

async 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; see fluree.Ledger.at().

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

Every change to subject, oldest first; see fluree.Ledger.history().

async log(limit: int | None = None) → list[Commit]

The ledger’s commits, newest first; see fluree.Ledger.log().

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

The facts one commit asserted and retracted; see fluree.Ledger.changes().

async branch(name: str) → Ledger

Create branch name from this ledger’s latest state and return it; see fluree.Ledger.branch().

async branches() → list[Branch]

Every branch of this ledger, by name; see fluree.Ledger.branches().

async 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; see fluree.Ledger.merge().

async merge_preview(source: str | Ledger, **options: Any) → MergePreview

See fluree.Ledger.merge_preview() for options.

async 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; see fluree.Ledger.rebase().

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

Undo one commit or several, in one new commit; see fluree.Ledger.revert().

async revert_preview(commits: CommitRef | list[CommitRef], **options: Any) → RevertPreview

See fluree.Ledger.revert_preview() for options.

async validate(shapes: Data | None = None, **options: Any) → ValidationReport

See fluree.Ledger.validate() for options.

async index_status() → IndexStatus

How far indexing has caught up with the ledger’s commits; see fluree.Ledger.index_status().

async index(*, timeout: float | None = None) → int

Index everything committed so far, waiting up to timeout seconds; see fluree.Ledger.index().

async reindex() → int

Rebuild the index from scratch from the commit history; see fluree.Ledger.reindex().

async 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; see fluree.Ledger.verify().

async 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; see fluree.Ledger.sweep().

async 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 ledger’s current data as RDF; see fluree.Ledger.export().

async 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; see fluree.Ledger.archive().

async context() → dict[str, Any] | None

The ledger’s default JSON-LD context (a property in the sync API).

async set_context(context: dict[str, Any]) → None

Replace the ledger’s default JSON-LD context; see fluree.Ledger.set_context().

async 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(); see fluree.Ledger.set_full_text().

async full_text() → FullText | None

The ledger’s full-text configuration, or None without one; see fluree.Ledger.full_text().

async info() → dict[str, Any]

Ledger metadata and statistics; see fluree.Ledger.info().

async graphs() → list[fluree.IRI]

The IRIs of the ledger’s named graphs, in the order they were first written to; see fluree.Ledger.graphs().

async drop_graph(graph: str) → Commit

Retract everything in named graph graph (its full IRI) in one commit; see fluree.Ledger.drop_graph().

class fluree.aio.Snapshot(ledger: Ledger, sync: fluree.Snapshot)

A ledger frozen at one point in time; see fluree.Snapshot.

property ledger: str

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

property t: int

The transaction this snapshot reflects; see fluree.Snapshot.t.

async 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, Cypher, or JSON-LD query; see fluree.Snapshot.query().

async 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; see fluree.Snapshot.select().

async 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; see fluree.Snapshot.profile().

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 fluree.Snapshot.stream().

async 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; see fluree.Snapshot.explain().

async 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 fluree.Snapshot.export().

async branch(name: str) → Ledger

Create branch name from this past state; see fluree.Snapshot.branch().

class fluree.aio.Transaction(sync: fluree.Transaction)

Writes committed together as one commit; see fluree.Transaction.

A write cannot be stopped part way, so cancelling a task that is waiting on one leaves it running on its worker thread. Leaving async with (and Ledger.transact() giving up) waits for that work to finish before rolling back, and the cancellation then propagates.

property committed: Commit | None

The Commit once the transaction has committed, else None.

async insert(data: Data, *, format: Format | None = None) → None

Stage an insert; see fluree.Transaction.insert().

async upsert(data: Data, *, format: Format | None = None) → None

Stage an upsert; see fluree.Transaction.upsert().

async insert_rows(rows: Any, **options: Any) → None

Stage Ledger.insert_rows(); see fluree.Transaction.insert_rows().

async upsert_rows(rows: Any, **options: Any) → None

Stage Ledger.upsert_rows(); see fluree.Transaction.upsert_rows().

async 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 fluree.Transaction.update().

async 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 fluree.Transaction.query().

async 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 fluree.Transaction.select().

async 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; see fluree.Transaction.explain().

async commit(*, message: str | None = None) → Commit

Commit the staged writes as one commit, recording message (by default the one the transaction was opened with); see fluree.Transaction.commit().

async rollback() → None

Discard the staged writes and close the transaction; see fluree.Transaction.rollback().

class fluree.aio.GraphSource(sync: fluree.GraphSource)

A graph source; see fluree.GraphSource.

property id: str

The source’s id, name:branch.

property name: str

The source’s name.

property branch: str

The source’s branch.

property kind: str

The kind of source, such as "iceberg", "delta" or "sql".

async 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 fluree.GraphSource.query().

async 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 fluree.GraphSource.select().

async 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; see fluree.GraphSource.materialize().

async drop() → None

Remove this graph source; see fluree.GraphSource.drop().

class fluree.aio.RowStream(start: collections.abc.Callable[[], fluree.RowStream])

Bases: collections.abc.AsyncIterator[Record]

Rows of a SELECT read as the query produces them; see fluree.RowStream. async with closes it at the end of the block, and so does cancelling the task reading it.

property columns: list[str] | None

The same as keys; see fluree.RowStream.columns.

async aclose() → None

Stop the query; the stream yields nothing more.