fluree.aio ========== .. py:module:: fluree.aio .. autoapi-nested-parse:: The :mod:`fluree` API for asyncio. The same classes and methods as :mod:`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 --------------- .. py:function:: connect(path: str | os.PathLike[str] | None = None, *, config: dict[str, Any] | None = None, indexing: bool = True) -> _Opening[Connection] Open a database; see :func:`fluree.connect`. Await it, or use it with ``async with`` to close the connection at the end of the block. .. py:class:: Connection(sync: fluree.Connection) A connection to a Fluree database; see :class:`fluree.Connection`. .. py:method:: close() -> None :async: Flush pending writes and release the database; see :meth:`fluree.Connection.close`. .. py:method:: exists(ledger: str) -> bool :async: Whether ``ledger`` exists (``ledger in conn`` for the sync API). .. py:method:: create(ledger: str, *, source: str | os.PathLike[str] | None = None) -> Ledger :async: Create a ledger, optionally bulk-loading RDF files into it; see :meth:`fluree.Connection.create`. .. py:method:: ledger(ledger: str) -> Ledger :async: An existing ledger; see :meth:`fluree.Connection.ledger`. .. py:method:: ledgers() -> list[str] :async: The ids of every ledger, as ``name:branch``; see :meth:`fluree.Connection.ledgers`. .. py:method:: query(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> Any :async: Run a query whose ``FROM`` names the ledgers to read; see :meth:`fluree.Connection.query`. .. py:method:: select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._results.Result :async: A SPARQL ``SELECT`` whose ``FROM`` names the ledgers, as for :meth:`query`, returning its :class:`Result`; see :meth:`fluree.Connection.select`. .. py:method:: profile(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._connection.QueryProfile :async: Run :meth:`query` and report the fuel and time it took; see :meth:`fluree.Connection.profile`. .. py:method:: map_iceberg(name: str, mapping: str | os.PathLike[str], **options: Any) -> GraphSource :async: Register Iceberg tables; see :meth:`fluree.Connection.map_iceberg`. .. py:method:: map_delta(name: str, mapping: str | os.PathLike[str], **options: Any) -> GraphSource :async: Register Delta tables; see :meth:`fluree.Connection.map_delta`. .. py:method:: map_sql(name: str, endpoint: str, mapping: str | os.PathLike[str], **options: Any) -> GraphSource :async: Register SQL tables; see :meth:`fluree.Connection.map_sql`. .. py:method:: graph_sources() -> list[GraphSource] :async: Every graph source, by id; see :meth:`fluree.Connection.graph_sources`. .. py:method:: graph_source(name: str) -> GraphSource :async: Graph source ``name`` (``"name"`` or ``"name:branch"``); see :meth:`fluree.Connection.graph_source`. .. py:method:: restore(path: str | os.PathLike[str], ledger: str) -> Ledger :async: Create ``ledger`` from a ``.flpack`` archive; see :meth:`fluree.Connection.restore`. .. py:method:: drop(ledger: str) -> None :async: Delete a ledger, every branch and all history (``"people"``), or one branch (``"people:dev"``); see :meth:`fluree.Connection.drop`. .. py:class:: Ledger(connection: Connection, sync: fluree.Ledger) A ledger; see :class:`fluree.Ledger`. .. py:property:: id :type: str The ledger's id, ``name:branch``. .. py:method:: 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 :meth:`fluree.Ledger.with_policy`. .. py:method:: insert(data: fluree._connection.Data, *, format: fluree._connection.Format | None = None, message: str | None = None) -> fluree._records.Commit :async: Add data; see :meth:`fluree.Ledger.insert`. .. py:method:: upsert(data: fluree._connection.Data, *, format: fluree._connection.Format | None = None, message: str | None = None) -> fluree._records.Commit :async: Add data, replacing existing values of the properties it sets; see :meth:`fluree.Ledger.upsert`. .. py:method:: insert_rows(rows: Any, **options: Any) -> fluree._records.Commit :async: See :meth:`fluree.Ledger.insert_rows`. .. py:method:: upsert_rows(rows: Any, **options: Any) -> fluree._records.Commit :async: See :meth:`fluree.Ledger.upsert_rows`. .. py:method:: update(transaction: str | dict[str, Any] | os.PathLike[str], parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, message: str | None = None, **kwparameters: Any) -> fluree._records.Commit :async: Apply a SPARQL UPDATE, a Cypher write, or a JSON-LD ``where``/``delete``/``insert``; see :meth:`fluree.Ledger.update`. .. py:method:: sync(data: fluree._connection.Data, *, graph: str | None = None, format: fluree._connection.Format | None = None, allow_empty: bool = False, dry_run: bool = False, message: str | None = None) -> fluree._records.Commit :async: Make a graph hold exactly ``data``, committing only the difference; see :meth:`fluree.Ledger.sync`. .. py:method:: transact(fn: collections.abc.Callable[..., collections.abc.Awaitable[T]], /, *args: Any, **kwargs: Any) -> T :async: Run ``await fn(txn, *args, **kwargs)`` in a :class:`Transaction` and commit it, running ``fn`` again if another commit lands first; see :meth:`fluree.Ledger.transact`. .. py:method:: transaction(*, message: str | None = None) -> _Opening[Transaction] Open a :class:`Transaction`. Await it, or use it with ``async with`` to commit on a clean exit and roll back on an exception. .. py:method:: query(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> Any :async: Query the latest state; see :meth:`fluree.Ledger.query`. .. py:method:: select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._results.Result :async: Run a ``SELECT`` on the latest state; see :meth:`fluree.Ledger.select`. .. py:method:: profile(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._connection.QueryProfile :async: Run :meth:`query` and report the fuel and time it took; see :meth:`fluree.Ledger.profile`. .. py:method:: stream(query: fluree._connection.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. .. py:method:: explain(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, **kwparameters: Any) -> dict[str, Any] :async: The plan the engine would run ``query`` with, without running it; see :meth:`fluree.Ledger.explain`. .. py:method:: snapshot() -> Snapshot :async: The latest state, frozen; see :meth:`fluree.Ledger.snapshot`. .. py:method:: at(t: int | None = None, *, time: datetime.datetime | str | None = None, commit: str | None = None) -> Snapshot :async: The ledger as it was at transaction ``t``, a wall-clock ``time``, or a ``commit``; see :meth:`fluree.Ledger.at`. .. py:method:: history(subject: str, predicate: str | None = None, *, from_t: int = 1, to_t: int | None = None) -> list[fluree._records.Change] :async: Every change to ``subject``, oldest first; see :meth:`fluree.Ledger.history`. .. py:method:: log(limit: int | None = None) -> list[fluree._records.Commit] :async: The ledger's commits, newest first; see :meth:`fluree.Ledger.log`. .. py:method:: changes(commit: fluree._connection.CommitRef) -> list[fluree._records.Change] :async: The facts one commit asserted and retracted; see :meth:`fluree.Ledger.changes`. .. py:method:: branch(name: str) -> Ledger :async: Create branch ``name`` from this ledger's latest state and return it; see :meth:`fluree.Ledger.branch`. .. py:method:: branches() -> list[fluree._records.Branch] :async: Every branch of this ledger, by name; see :meth:`fluree.Ledger.branches`. .. py:method:: merge(source: str | Ledger, *, strategy: fluree._connection.MergeStrategy = 'take-both') -> fluree._records.MergeResult :async: Bring the commits of branch ``source`` (a name like ``"dev"``, or its :class:`Ledger`) into this branch; see :meth:`fluree.Ledger.merge`. .. py:method:: merge_preview(source: str | Ledger, **options: Any) -> fluree._records.MergePreview :async: See :meth:`fluree.Ledger.merge_preview` for ``options``. .. py:method:: rebase(*, strategy: fluree._connection.RebaseStrategy = 'take-both') -> fluree._records.RebaseResult :async: 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 :meth:`fluree.Ledger.rebase`. .. py:method:: revert(commits: fluree._connection.CommitRef | list[fluree._connection.CommitRef], *, strategy: fluree._connection.RevertStrategy = 'abort') -> fluree._records.RevertResult :async: Undo one commit or several, in one new commit; see :meth:`fluree.Ledger.revert`. .. py:method:: revert_preview(commits: fluree._connection.CommitRef | list[fluree._connection.CommitRef], **options: Any) -> fluree._records.RevertPreview :async: See :meth:`fluree.Ledger.revert_preview` for ``options``. .. py:method:: validate(shapes: fluree._connection.Data | None = None, **options: Any) -> fluree._records.ValidationReport :async: See :meth:`fluree.Ledger.validate` for ``options``. .. py:method:: index_status() -> fluree._records.IndexStatus :async: How far indexing has caught up with the ledger's commits; see :meth:`fluree.Ledger.index_status`. .. py:method:: index(*, timeout: float | None = None) -> int :async: Index everything committed so far, waiting up to ``timeout`` seconds; see :meth:`fluree.Ledger.index`. .. py:method:: reindex() -> int :async: Rebuild the index from scratch from the commit history; see :meth:`fluree.Ledger.reindex`. .. py:method:: verify(*, max_commits: int | None = None) -> fluree._records.VerifyReport :async: Check that every commit, back to the first (or the last ``max_commits``), and the index root are present and readable; see :meth:`fluree.Ledger.verify`. .. py:method:: sweep(*, dry_run: bool = False) -> fluree._records.SweepResult :async: Delete index files that no index of any branch of this ledger references any more — left behind as indexing replaces old index files; see :meth:`fluree.Ledger.sweep`. .. py:method:: export(path: str | os.PathLike[str] | None = None, *, format: fluree._connection.ExportFormat | None = None, graph: str | None = None, all_graphs: bool = False, context: dict[str, Any] | None = None) -> str | None :async: Export the ledger's current data as RDF; see :meth:`fluree.Ledger.export`. .. py:method:: archive(path: str | os.PathLike[str], *, include_indexes: bool = True) -> None :async: Write the whole ledger, every commit, to a ``.flpack`` archive that :meth:`Connection.restore` loads back; see :meth:`fluree.Ledger.archive`. .. py:method:: context() -> dict[str, Any] | None :async: The ledger's default JSON-LD context (a property in the sync API). .. py:method:: set_context(context: dict[str, Any]) -> None :async: Replace the ledger's default JSON-LD context; see :meth:`fluree.Ledger.set_context`. .. py:method:: set_full_text(properties: collections.abc.Iterable[str], *, language: str = 'en', reindex: bool = True) -> None :async: Make the plain-string values of ``properties`` searchable with ``fulltext()``; see :meth:`fluree.Ledger.set_full_text`. .. py:method:: full_text() -> fluree._records.FullText | None :async: The ledger's full-text configuration, or ``None`` without one; see :meth:`fluree.Ledger.full_text`. .. py:method:: info() -> dict[str, Any] :async: Ledger metadata and statistics; see :meth:`fluree.Ledger.info`. .. py:method:: graphs() -> list[fluree.IRI] :async: The IRIs of the ledger's named graphs, in the order they were first written to; see :meth:`fluree.Ledger.graphs`. .. py:method:: drop_graph(graph: str) -> fluree._records.Commit :async: Retract everything in named graph ``graph`` (its full IRI) in one commit; see :meth:`fluree.Ledger.drop_graph`. .. py:class:: Snapshot(ledger: Ledger, sync: fluree.Snapshot) A ledger frozen at one point in time; see :class:`fluree.Snapshot`. .. py:property:: ledger :type: str The id of the ledger this is a snapshot of, ``name:branch``. .. py:property:: t :type: int The transaction this snapshot reflects; see :attr:`fluree.Snapshot.t`. .. py:method:: query(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> Any :async: Run a SPARQL, Cypher, or JSON-LD query; see :meth:`fluree.Snapshot.query`. .. py:method:: select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._results.Result :async: Run a SPARQL ``SELECT`` or a Cypher query and return its :class:`Result` — :meth:`query` for the queries whose result is a table, typed as one; see :meth:`fluree.Snapshot.select`. .. py:method:: profile(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._connection.QueryProfile :async: Run :meth:`query` and report the fuel and time it took; see :meth:`fluree.Snapshot.profile`. .. py:method:: stream(query: fluree._connection.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 :meth:`fluree.Snapshot.stream`. .. py:method:: explain(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, **kwparameters: Any) -> dict[str, Any] :async: The plan the engine would run ``query`` with, without running it; see :meth:`fluree.Snapshot.explain`. .. py:method:: export(path: str | os.PathLike[str] | None = None, *, format: fluree._connection.ExportFormat | None = None, graph: str | None = None, all_graphs: bool = False, context: dict[str, Any] | None = None) -> str | None :async: Export the data as of this snapshot; see :meth:`fluree.Snapshot.export`. .. py:method:: branch(name: str) -> Ledger :async: Create branch ``name`` from this past state; see :meth:`fluree.Snapshot.branch`. .. py:class:: Transaction(sync: fluree.Transaction) Writes committed together as one commit; see :class:`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 :meth:`Ledger.transact` giving up) waits for that work to finish before rolling back, and the cancellation then propagates. .. py:property:: committed :type: fluree._records.Commit | None The :class:`Commit` once the transaction has committed, else ``None``. .. py:method:: insert(data: fluree._connection.Data, *, format: fluree._connection.Format | None = None) -> None :async: Stage an insert; see :meth:`fluree.Transaction.insert`. .. py:method:: upsert(data: fluree._connection.Data, *, format: fluree._connection.Format | None = None) -> None :async: Stage an upsert; see :meth:`fluree.Transaction.upsert`. .. py:method:: insert_rows(rows: Any, **options: Any) -> None :async: Stage :meth:`Ledger.insert_rows`; see :meth:`fluree.Transaction.insert_rows`. .. py:method:: upsert_rows(rows: Any, **options: Any) -> None :async: Stage :meth:`Ledger.upsert_rows`; see :meth:`fluree.Transaction.upsert_rows`. .. py:method:: update(transaction: str | dict[str, Any] | os.PathLike[str], parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, **kwparameters: Any) -> fluree._results.Result | None :async: Stage an update — SPARQL UPDATE, a Cypher write, or JSON-LD; see :meth:`fluree.Transaction.update`. .. py:method:: query(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> Any :async: Query the staged state; see :meth:`fluree.Transaction.query`. .. py:method:: select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._results.Result :async: Run a ``SELECT`` on the staged state; see :meth:`fluree.Transaction.select`. .. py:method:: explain(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, **kwparameters: Any) -> dict[str, Any] :async: The plan ``query`` would run with over the staged state; see :meth:`fluree.Transaction.explain`. .. py:method:: commit(*, message: str | None = None) -> fluree._records.Commit :async: Commit the staged writes as one commit, recording ``message`` (by default the one the transaction was opened with); see :meth:`fluree.Transaction.commit`. .. py:method:: rollback() -> None :async: Discard the staged writes and close the transaction; see :meth:`fluree.Transaction.rollback`. .. py:class:: GraphSource(sync: fluree.GraphSource) A graph source; see :class:`fluree.GraphSource`. .. py:property:: id :type: str The source's id, ``name:branch``. .. py:property:: name :type: str The source's name. .. py:property:: branch :type: str The source's branch. .. py:property:: kind :type: str The kind of source, such as ``"iceberg"``, ``"delta"`` or ``"sql"``. .. py:method:: query(query: fluree._connection.Query, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.Language | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> Any :async: Run a SPARQL or JSON-LD query of this source; see :meth:`fluree.GraphSource.query`. .. py:method:: select(query: str, parameters: collections.abc.Mapping[str, Any] | None = None, *, language: fluree._connection.SelectLanguage | None = None, max_fuel: float | None = None, timeout: float | None = None, **kwparameters: Any) -> fluree._results.Result :async: Run a SPARQL ``SELECT`` of this source; see :meth:`fluree.GraphSource.select`. .. py:method:: materialize(into: str, *, full: bool = False) -> fluree._sources.MaterializeResult :async: Copy this source's rows into ledger ``into`` (created if missing) as ordinary ledger data, so they gain history, policy and indexes; see :meth:`fluree.GraphSource.materialize`. .. py:method:: drop() -> None :async: Remove this graph source; see :meth:`fluree.GraphSource.drop`. .. py:class:: RowStream(start: collections.abc.Callable[[], fluree.RowStream]) Bases: :py:obj:`collections.abc.AsyncIterator`\ [\ :py:obj:`fluree._results.Record`\ ] Rows of a SELECT read as the query produces them; see :class:`fluree.RowStream`. ``async with`` closes it at the end of the block, and so does cancelling the task reading it. .. py:property:: columns :type: list[str] | None The same as :meth:`keys`; see :attr:`fluree.RowStream.columns`. .. py:method:: aclose() -> None :async: Stop the query; the stream yields nothing more.