The clio_cte_core_ext module.

Connecting

Set environment variables before the import; the client reads them once when it connects.

import os
os.environ["CLIO_WITH_RUNTIME"] = "0"    # attach to a running clio_run
os.environ["CLIO_CTE_POOL"] = "564.0"    # only needed for SemanticSearch

import clio_cte_core_ext as cte
cte.clio_init(cte.RuntimeMode.kClient, False)
cte.initialize_cte("", cte.PoolQuery.Dynamic())
client = cte.get_cte_client()

• clio_init(mode, default_with_runtime=False, is_restart=False) -> bool. CLIO_WITH_RUNTIME overrides the second argument. Waits up to CLIO_WAIT_SERVER seconds (default 30) for the runtime.
• initialize_cte(config_path, pool_query) -> bool. Both arguments are required; pass "" and PoolQuery.Dynamic() for the defaults.
• get_cte_client() -> Client: a copy of the shared client. Client() and Client(pool_id) construct one directly.

Tag

Client: blocking calls

Patterns must match the whole name (std::regex_match), so use .*part.* for a substring. SemanticSearch ranks every blob matching both patterns, including blobs with no query word (score 0). For files in a mount, use "[0-9]+" as the blob pattern to search pages and skip each file's attribute blob.

Client: asynchronous calls

Each returns a Future. Submission runs without the GIL and may block while shared memory is allocated.

Future

• wait(max_sec=-1): blocks until done; returns the return code, or None if max_sec (when positive) elapses first.
• done(): True when the task has completed.
• result(max_sec=-1): blocks, then returns the value in the table above. Raises TimeoutError if max_sec elapses.

Deferred puts

• AsyncPutBlobDefer(tag_id, blob_name, data, off=0, max_inflight_bytes=0) -> int: stages a copy of data in shared memory and returns 0 when submitted. data is free to reuse as soon as the call returns. When shared memory is exhausted the call waits for the oldest puts; max_inflight_bytes adds an explicit ceiling (0 means none).
• AwaitPutsUntilSpace(max_inflight_bytes) -> int: waits until at most that many payload bytes are in flight (0 drains all) and returns the bytes still in flight.
• DeferErrorCount() -> int: deferred puts that completed with a non-zero return code. The count is sticky.

Supporting types

• UniqueId(major, minor): the tag id, with major_, minor_, IsNull(), ToU64() and the static GetNull().
• PoolQuery.Broadcast(), PoolQuery.Dynamic(), PoolQuery.Local(parallelism=32).
• SemanticSearchResult: tag_id, blob_name, score. TemporalSearchResult: tag_id, blob_name, last_modified.
• CteTelemetry: op_ (a CteOp), off_, size_, tag_id_, mod_time_, read_time_, logical_time_.
• CapacityInfo: total_capacity, remaining_capacity. TargetInfo: target_score, remaining_space, bytes_read, bytes_written, ops_read, ops_written.
• Enums: RuntimeMode (kClient, kServer, kRuntime), BdevType (kFile, kRam), CteOp (kPutBlob, kGetBlob, kDelBlob, kGetOrCreateTag, kDelTag, kGetTagSize).

Example

tag = cte.Tag("experiment/run1")
tag.PutBlob("note0", b"Arctic sea ice extent fell to a record low.", 0)
tag.PutBlob("note1", b"Atmospheric CO2 passed 420 ppm.", 0)

for tag_name, blob_name in client.BlobQuery("experiment/.*", "note.*"):
    print(tag_name, blob_name)

for hit in client.SemanticSearch("experiment/.*", ".*", "sea ice", k=5):
    print(hit.blob_name, hit.score)

tag.ReorganizeBlob("note1", 0.0)

futs = [client.AsyncGetBlob(tag.GetTagId(), n, tag.GetBlobSize(n), 0) for n in ("note0", "note1")]
payloads = [f.result() for f in futs]

client.DelBlob(tag.GetTagId(), "note0")