Python API¶

Bases: _LaminDBFlowOperator

Start the LaminDB flow run for this DAG run, in a virtualenv.

Creates (or restarts, on retry) a Run of the DAG file's Transform and tags it with the DAG run so downstream LaminDB steps can find it. Wrap the steps in with init >> finish: to wire them, or let steps with auto_flow=True add and wire one. Returns the flow run uid.

Marked as an Airflow setup task by default, pairing with the teardown LaminDBFlowFinishOperator; pass is_setup=False to opt out. Accepts every PythonVirtualenvOperator argument except python_callable, op_args and op_kwargs.

129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

def __init__(
    self,
    *,
    task_id: str = "lamindb_flow_init",
    is_setup: bool = True,
    lamindb_conn_id: str | None = LaminDBHook.default_conn_name,
    lamindb_instance: str | None = None,
    lamindb_version: str | None = None,
    lamindb_cache_dir: str | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(
        task_id=task_id,
        lamindb_conn_id=lamindb_conn_id,
        lamindb_instance=lamindb_instance,
        lamindb_version=lamindb_version,
        lamindb_cache_dir=lamindb_cache_dir,
        **kwargs,
    )
    if is_setup:
        self.as_setup()

Bases: _LaminDBFlowOperator

Close the LaminDB flow run with the DAG run's outcome, in a virtualenv.

Marked as an Airflow teardown task by default: it runs after every other task (trigger_rule=all_done_setup_success) and is ignored when Airflow decides the DAG run state, so a failed step still fails the DAG run. Marks the flow run errored if any other task in the DAG run failed. Pass is_teardown=False together with your own trigger_rule to opt out. Accepts every PythonVirtualenvOperator argument except python_callable, op_args and op_kwargs.

180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202

def __init__(
    self,
    *,
    task_id: str = "lamindb_flow_finish",
    is_teardown: bool = True,
    lamindb_conn_id: str | None = LaminDBHook.default_conn_name,
    lamindb_instance: str | None = None,
    lamindb_version: str | None = None,
    lamindb_cache_dir: str | None = None,
    **kwargs: Any,
) -> None:
    if not is_teardown:
        kwargs.setdefault("trigger_rule", TriggerRule.ALL_DONE)
    super().__init__(
        task_id=task_id,
        lamindb_conn_id=lamindb_conn_id,
        lamindb_instance=lamindb_instance,
        lamindb_version=lamindb_version,
        lamindb_cache_dir=lamindb_cache_dir,
        **kwargs,
    )
    if is_teardown:
        self.as_teardown()

Bases: LaminDBDbWriteEventTrigger

Fire an event for every record of a LaminDB registry that is created, updated or deleted.

Use it in an AssetWatcher to schedule DAGs on LaminDB changes. Each event payload contains the event (created/updated/deleted), the current record (None for hard deletes), the previous values logged with the write, the changed_fields of updates, changed_by and the underlying dbwrite log entry.

With branch set (main by default), events are relative to that branch: created also covers records that arrive on the branch by merging a contribution branch or by restoring them from the trash or archive, and deleted also covers records moved off the branch, for example to the trash via record.delete(). For registries without branches, branch is ignored.

Parameters:

• registry (str) –

  Registry to watch, e.g. LaminDBRegistry.COLLECTION, "core.run", "core.record" or "bionty.celltype".

• events (str | Sequence[str], default: ('created',) ) –

  Events to report, any of created, updated and deleted.

• branch (str | int | None, default: 'main' ) –

  Name or id of the branch to watch; None reports raw inserts, updates and deletes on all branches.

• filter (FilterLike | None, default: None ) –

  Filter the record must match, built with F, e.g. F(ArtifactField.KEY).startswith("raw/"), or a LaminHub REST filter such as {"key": {"startswith": "raw/"}}. It's evaluated against the current record; for hard deletes against the deleted row, which only supports direct fields.

• changed_fields (Sequence[str] | None, default: None ) –

  Only report updates that change at least one of these fields, e.g. [ArtifactField.KEY].

  Records that hold trigger cursors (see cursor_store) are never reported. They are recognized by the registry and reference_type of this trigger's own RecordCursorStore (by default core.record and airflow_trigger_cursor), so give triggers that store cursors in the registry they watch the same settings.

111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

def __init__(
    self,
    registry: str,
    *,
    events: str | Sequence[str] = ("created",),
    branch: str | int | None = "main",
    filter: FilterLike | None = None,
    changed_fields: Sequence[str] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.registry: str = plain_value(registry)
    self.events = _normalize_events(events)
    self.branch = branch
    self.filter = normalize_filter(filter)
    self.changed_fields = normalize_str_list(changed_fields) if changed_fields else None
    if "deleted" in self.events and not is_local_filter(self._record_filter()):
        raise ValueError(
            "'deleted' events only support filters on direct fields of the record "
            "(no relation or JSON paths), because deleted records can't be queried anymore."
        )
    if (
        isinstance(self.cursor_store, RecordCursorStore)
        and self.cursor_store.reference_type is None
        and normalize_registry_name(self.cursor_store.registry) == normalize_registry_name(self.registry)
    ):
        raise ValueError(
            "This trigger stores its cursor in the registry it watches, so it needs a "
            "reference_type to recognize its own saves instead of reacting to them."
        )
    self._registry: Registry | None = None
    self._branch_id: int | None = None
    # how cursor records look: this trigger's own record cursor store, or the defaults
    self._cursor_settings = (
        self.cursor_store if isinstance(self.cursor_store, RecordCursorStore) else RecordCursorStore()
    )
    self._cursor_record_ids: set[int] = set()
    self._watches_cursor_records = False

Bases: LaminDBRecordEventTrigger

Fire an event for every LaminDB artifact that is created, updated or deleted.

A specialization of LaminDBRecordEventTrigger for core.artifact. By default, the created event fires only once the artifact's upload to storage has completed (wait_for_upload), so that downstream tasks can load it right away. Artifacts that LaminDB creates internally (run logs, environments) are ignored unless include_internal is set.

Parameters:

• key (str | None, default: None ) –

  Only artifacts with this exact key.

• key_prefix (str | None, default: None ) –

  Only artifacts whose key starts with this prefix, e.g. "raw/".

• suffix (str | Sequence[str] | None, default: None ) –

  Only artifacts with this suffix (or any of these suffixes), e.g. ".parquet".

• kind (str | Sequence[str] | None, default: None ) –

  Only artifacts of this kind (or any of these kinds), e.g. ArtifactKind.DATASET.

• include_internal (bool, default: False ) –

  Also report artifacts LaminDB creates internally.

• wait_for_upload (bool, default: True ) –

  Report created only once the upload to storage has completed and skip updates of artifacts that are being uploaded.

• events –

  See LaminDBRecordEventTrigger.

• branch –

  See LaminDBRecordEventTrigger.

• filter –

  Additional filter, see LaminDBRecordEventTrigger.

• changed_fields –

  See LaminDBRecordEventTrigger.

346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364

def __init__(
    self,
    *,
    key: str | None = None,
    key_prefix: str | None = None,
    suffix: str | Sequence[str] | None = None,
    kind: str | Sequence[str] | None = None,
    include_internal: bool = False,
    wait_for_upload: bool = True,
    **kwargs: Any,
) -> None:
    kwargs.pop("registry", None)
    self.key = key
    self.key_prefix = key_prefix
    self.suffix = normalize_str_list(suffix)
    self.kind = normalize_str_list(kind)
    self.include_internal = include_internal
    self.wait_for_upload = wait_for_upload
    super().__init__("core.artifact", **kwargs)

Bases: LaminDBDbWriteEventTrigger

Fire an event whenever the status of a LaminDB branch changes.

Branch statuses are standalone, draft and review (a Change Request is open), and merged and closed. The event payload contains from_status, to_status, the branch (with its current status), changed_by and the underlying dbwrite entry.

Parameters:

• to_status (str | Sequence[str] | None, default: None ) –

  Only report transitions to this status (or any of these statuses).

• from_status (str | Sequence[str] | None, default: None ) –

  Only report transitions from this status (or any of these statuses).

• branch_name (str | Sequence[str] | None, default: None ) –

  Only report branches whose name matches this glob pattern (or any of these patterns), e.g. "release-*".

80
81
82
83
84
85
86
87
88
89
90
91

def __init__(
    self,
    *,
    to_status: str | Sequence[str] | None = None,
    from_status: str | Sequence[str] | None = None,
    branch_name: str | Sequence[str] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.to_status = normalize_statuses(to_status, name="to_status")
    self.from_status = normalize_statuses(from_status, name="from_status")
    self.branch_name = normalize_str_list(branch_name)

Bases: LaminDBDbWriteEventTrigger

Fire an event whenever a comment or readme is added to a LaminDB branch.

Comments and readmes are core.branchblock records; a new readme version is a new block. The event payload contains the kind, the block (including its Markdown content), the branch, changed_by and the underlying dbwrite entry.

Parameters:

• kinds (str | Sequence[str], default: BLOCK_KINDS ) –

  Block kinds to report, comment and/or readme.

• branch_name (str | Sequence[str] | None, default: None ) –

  Only report blocks on branches whose name matches this glob pattern (or any of these patterns).

185
186
187
188
189
190
191
192
193
194
195
196
197
198
199

def __init__(
    self,
    *,
    kinds: str | Sequence[str] = BLOCK_KINDS,
    branch_name: str | Sequence[str] | None = None,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.kinds = normalize_str_list(kinds) or []
    invalid = [kind for kind in self.kinds if kind not in BLOCK_KINDS]
    if invalid or not self.kinds:
        raise ValueError(
            f"Invalid block kinds {self.kinds}; expected a non-empty subset of {BLOCK_KINDS}"
        )
    self.branch_name = normalize_str_list(branch_name)

Bases: BaseEventTrigger

Base class for event-driven scheduling triggers that watch the LaminHub database write log.

LaminHub logs every INSERT, UPDATE and DELETE on a LaminDB instance's database in the hubmodule.dbwrite registry ("Changes → Database writes" in the LaminHub UI). Subclasses define which writes to poll and how to turn them into trigger events.

The trigger polls writes with an id greater than its cursor, and stores the cursor under state_key so that a restarted triggerer resumes where it left off. The cursor is committed at least commit_delay seconds after the events were yielded, which gives Airflow time to persist them: events are delivered at least once. It is only committed when the trigger processed new writes. On the first start the trigger only reports writes that happen from then on.

Parameters:

• lamindb_conn_id (str | None, default: default_conn_name ) –

  Airflow connection of type lamindb.

• instance (str | None, default: None ) –

  LaminDB instance owner/name; overrides the connection's instance.

• poll_interval (float, default: 30.0 ) –

  Seconds between polls of the write log.

• batch_size (int, default: MAX_PAGE_SIZE ) –

  Number of writes fetched per request (at most 200).

• max_batches_per_poll (int, default: 10 ) –

  Maximum number of batches processed per poll cycle.

• cursor_store (CursorStore | Mapping[str, Any] | None, default: None ) –

  Where and how to store the cursor: AssetCursorStore (default) in the asset state store of the AssetWatcher's assets, or RecordCursorStore in a record of the watched instance. Both also set the commit_delay.

49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

def __init__(
    self,
    *,
    lamindb_conn_id: str | None = LaminDBHook.default_conn_name,
    instance: str | None = None,
    poll_interval: float = 30.0,
    batch_size: int = MAX_PAGE_SIZE,
    max_batches_per_poll: int = 10,
    cursor_store: CursorStore | Mapping[str, Any] | None = None,
) -> None:
    super().__init__()
    if poll_interval <= 0:
        raise ValueError("poll_interval must be positive")
    if not 1 <= batch_size <= MAX_PAGE_SIZE:
        raise ValueError(f"batch_size must be between 1 and {MAX_PAGE_SIZE}")
    if max_batches_per_poll < 1:
        raise ValueError("max_batches_per_poll must be at least 1")
    self.lamindb_conn_id = lamindb_conn_id
    self.instance = instance
    self.poll_interval = poll_interval
    self.batch_size = batch_size
    self.max_batches_per_poll = max_batches_per_poll
    self.cursor_store = CursorStore.coerce(cursor_store)
    self._users: dict[int, dict[str, Any]] = {}

state_key: str

Where and how an event trigger keeps its cursor; use a subclass.

Parameters:

• commit_delay (float, default: DEFAULT_COMMIT_DELAY ) –

  Minimum seconds between yielding events and committing the cursor past them, which gives Airflow time to persist the events. Longer delays save less often, but redeliver more events after a crash.

coerce(value: CursorStore | Mapping[str, Any] | None) -> CursorStore

65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81

@classmethod
def coerce(cls, value: CursorStore | Mapping[str, Any] | None) -> CursorStore:
    """Accept a store, its ``serialize`` form (as in serialized trigger kwargs) or None."""
    if value is None:
        return AssetCursorStore()
    if isinstance(value, CursorStore):
        return value
    if not isinstance(value, Mapping):
        raise TypeError(f"cursor_store must be a CursorStore, e.g. RecordCursorStore(), got {value!r}")
    settings = dict(value)
    store_type = CursorStore._types.get(settings.pop("type", None))
    if store_type is None:
        raise ValueError(f"cursor_store type must be one of {sorted(CursorStore._types)}, got {value!r}")
    unknown = set(settings) - {f.name for f in fields(store_type)}
    if unknown:
        raise ValueError(f"Unknown {store_type.__name__} settings: {sorted(unknown)}")
    return store_type(**settings)

open(
    *, hook: LaminDBHook, asset_state_store: Any, key: str, description: str
) -> CursorBackend

86
87
88

def open(self, *, hook: LaminDBHook, asset_state_store: Any, key: str, description: str) -> CursorBackend:
    """The backend that loads and saves the cursor under ``key`` while the trigger runs."""
    raise NotImplementedError

Bases: CursorStore

Keep the cursor in the asset state store of the AssetWatcher's assets.

The cursor lives in Airflow's metadata database next to the asset events. A trigger outside an AssetWatcher has no asset state store and keeps its cursor in memory only.

Bases: CursorStore

Keep the cursor in a record of the watched instance; needs write access to the instance.

The defaults keep each trigger's cursor in its own core.record on main, grouped under the record type Airflow trigger cursors. The record and its type are created on the first save. Deleting the record resets the cursor when the trigger restarts; a running trigger creates the record again on its next save.

Parameters:

• registry (str, default: 'core.record' ) –

  Registry of the cursor records. It needs the fields name, branch, space and field, plus reference_type and type/is_type unless those are disabled below.

• field (str, default: 'extra_data' ) –

  JSON field that holds the cursor, {"dbwrite_id": ..., "instance_id": ...}.

• record_type (str | None, default: CURSOR_RECORD_TYPE_NAME ) –

  Name of the record type (is_type=True) that groups the cursor records; None for no type, for registries without types.

• reference_type (str | None, default: CURSOR_REFERENCE_TYPE ) –

  Marker in reference_type that identifies cursor records, so that record triggers on registry can skip them; None for no marker. Then such triggers report the cursor records like any other record. A record trigger watching registry itself needs a marker, so it rejects None.

• branch (str | int, default: 'main' ) –

  Branch (name or id) of the cursor records and their type.

• space (str | int | None, default: None ) –

  Space (name or id) of the cursor records and their type, e.g. a restricted space that only the Airflow API key can write to; None for LaminHub's default space.

• name (str | None, default: None ) –

  Name of the cursor record; None for the trigger's state_key. A fixed name must be unique per trigger, or triggers overwrite each other's cursors.

• description (str | None, default: None ) –

  Description of the cursor record; None to describe the trigger.

• commit_delay (float, default: DEFAULT_COMMIT_DELAY ) –

  See CursorStore.

Bases: BaseSensorOperator

Wait until a LaminDB branch reaches a status, e.g. until its Change Request is merged.

Returns the branch (with its status), which is pushed to XCom. Fails without retries if the branch reaches one of the failure_status statuses (by default, when the Change Request is closed without merging). Network errors, rate limiting and LaminHub server errors don't end the wait; the sensor checks again later.

For more information on how to use this sensor, take a look at the guide.

Parameters:

• branch (str | int) –

  Branch name or id.

• target_status (str | Sequence[str], default: 'merged' ) –

  Status (or statuses) to wait for, any of standalone, draft, review, merged and closed.

• failure_status (str | Sequence[str], default: ('closed',) ) –

  Status (or statuses) that fail the sensor.

• lamindb_conn_id (str | None, default: default_conn_name ) –

  Airflow connection of type lamindb.

• instance (str | None, default: None ) –

  LaminDB instance owner/name; overrides the connection's instance.

• deferrable (bool, default: getboolean('operators', 'default_deferrable', fallback=False) ) –

  Wait in the triggerer instead of occupying a worker slot.

45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

def __init__(
    self,
    *,
    branch: str | int,
    target_status: str | Sequence[str] = "merged",
    failure_status: str | Sequence[str] = ("closed",),
    lamindb_conn_id: str | None = LaminDBHook.default_conn_name,
    instance: str | None = None,
    deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.branch = branch
    self.target_status = normalize_statuses(target_status, name="target_status") or []
    self.failure_status = normalize_statuses(failure_status, name="failure_status") or []
    if not self.target_status:
        raise ValueError("target_status must not be empty")
    if overlap := set(self.target_status) & set(self.failure_status):
        raise ValueError(f"Statuses {sorted(overlap)} are both target and failure statuses")
    self.lamindb_conn_id = lamindb_conn_id
    self.instance = instance
    self.deferrable = deferrable

Bases: BaseSensorOperator

Wait until records matching a filter exist in a LaminDB registry.

Returns the matching records (newest first, at most limit), which are pushed to XCom. Network errors, rate limiting and LaminHub server errors don't end the wait; the sensor checks again later.

For more information on how to use this sensor, take a look at the guide.

Parameters:

• registry (str) –

  Registry to query, e.g. LaminDBRegistry.RUN, "core.artifact" or "bionty.celltype".

• filter (FilterLike | None, default: None ) –

  Filter built with F, e.g. F(ArtifactField.KEY) == "raw/sample1.fastq.gz", or a LaminHub REST filter such as {"key": {"eq": "raw/sample1.fastq.gz"}}. It's validated when the DAG is parsed.

• branch (str | int | None, default: 'main' ) –

  Only records on this branch (name or id); None for all branches and for registries without branches.

• min_count (int, default: 1 ) –

  Number of matching records to wait for.

• limit (int, default: 100 ) –

  Maximum number of records returned.

• wait_for_upload (bool | None, default: None ) –

  Ignore artifacts whose upload to storage is still in progress.

• lamindb_conn_id (str | None, default: default_conn_name ) –

  Airflow connection of type lamindb.

• instance (str | None, default: None ) –

  LaminDB instance owner/name; overrides the connection's instance.

• deferrable (bool, default: getboolean('operators', 'default_deferrable', fallback=False) ) –

  Wait in the triggerer instead of occupying a worker slot.

53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

def __init__(
    self,
    *,
    registry: str,
    filter: FilterLike | None = None,
    branch: str | int | None = "main",
    min_count: int = 1,
    limit: int = 100,
    wait_for_upload: bool | None = None,
    lamindb_conn_id: str | None = LaminDBHook.default_conn_name,
    instance: str | None = None,
    deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    if min_count < 1 or limit < min_count:
        raise ValueError("min_count must be at least 1 and not larger than limit")
    self.registry: str = plain_value(registry)
    self.filter = normalize_filter(filter)
    self.branch = branch
    self.min_count = min_count
    self.limit = limit
    self.wait_for_upload = wait_for_upload
    self.lamindb_conn_id = lamindb_conn_id
    self.instance = instance
    self.deferrable = deferrable

Bases: LaminDBRecordSensor

Wait until LaminDB artifacts matching some criteria exist and their upload has completed.

Returns the matching artifacts (newest first, at most limit), which are pushed to XCom.

For more information on how to use this sensor, take a look at the guide.

Parameters:

• key (str | None, default: None ) –

  Artifact key, e.g. "raw/sample1.fastq.gz".

• key_prefix (str | None, default: None ) –

  Key prefix, e.g. "raw/".

• suffix (str | Sequence[str] | None, default: None ) –

  Suffix (or suffixes), e.g. ".parquet".

• kind (str | Sequence[str] | None, default: None ) –

  Artifact kind (or kinds), e.g. ArtifactKind.DATASET.

• include_internal (bool, default: False ) –

  Also consider artifacts LaminDB creates internally (run logs, ...).

• filter –

  Additional filter, see LaminDBRecordSensor.

• wait_for_upload (bool, default: True ) –

  Ignore artifacts whose upload is still in progress (default).

160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

def __init__(
    self,
    *,
    key: str | None = None,
    key_prefix: str | None = None,
    suffix: str | Sequence[str] | None = None,
    kind: str | Sequence[str] | None = None,
    include_internal: bool = False,
    wait_for_upload: bool = True,
    **kwargs: Any,
) -> None:
    kwargs.pop("registry", None)
    super().__init__(registry="core.artifact", wait_for_upload=wait_for_upload, **kwargs)
    self.key = key
    self.key_prefix = key_prefix
    self.suffix = normalize_str_list(suffix)
    self.kind = normalize_str_list(kind)
    self.include_internal = include_internal

Bases: BaseHook

Interact with LaminDB instances hosted on LaminHub via the LaminHub REST API.

The connection's password is a Lamin API key; it is exchanged for a short-lived access token that is refreshed automatically. Without an API key, requests are anonymous, which works for public instances. The instance (owner/name) is read from the connection's instance extra and can be overridden per hook. Instances on lamindb older than 2.8.1 are rejected with LaminDBVersionError.

Parameters:

• lamindb_conn_id (str | None, default: default_conn_name ) –

  Airflow connection of type lamindb. None means anonymous access with default settings, which requires instance to be passed.

• instance (str | None, default: None ) –

  LaminDB instance slug owner/name. Overrides the connection's instance.

• timeout (float, default: 30.0 ) –

  Timeout for HTTP requests in seconds.

• retries (int, default: 3 ) –

  Number of retries for connection errors and retryable HTTP status codes (429 and 5xx).

• retry_backoff (float, default: 1.0 ) –

  Base delay in seconds for the exponential retry backoff.

190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212

def __init__(
    self,
    lamindb_conn_id: str | None = default_conn_name,
    instance: str | None = None,
    *,
    timeout: float = 30.0,
    retries: int = 3,
    retry_backoff: float = 1.0,
    **kwargs: Any,
) -> None:
    super().__init__(**kwargs)
    self.lamindb_conn_id = lamindb_conn_id
    self.instance_slug = instance
    self.timeout = timeout
    self.retries = retries
    self.retry_backoff = retry_backoff
    self._config: _HubConfig | None = None
    self._access_token: str | None = None
    self._access_token_expiry = 0.0
    self._instance: LaminDBInstance | None = None
    self._schema: dict[str, Any] | None = None
    self._client: httpx.Client | None = None
    self._async_client: httpx.AsyncClient | None = None

get_conn() -> Client

252
253
254
255
256

def get_conn(self) -> httpx.Client:
    """Return the (lazily created) synchronous HTTP client."""
    if self._client is None:
        self._client = httpx.Client(timeout=self.timeout, headers={"User-Agent": _USER_AGENT})
    return self._client

get_async_conn() -> AsyncClient

258
259
260
261
262

def get_async_conn(self) -> httpx.AsyncClient:
    """Return the (lazily created) asynchronous HTTP client."""
    if self._async_client is None:
        self._async_client = httpx.AsyncClient(timeout=self.timeout, headers={"User-Agent": _USER_AGENT})
    return self._async_client

get_instance_slug() -> str

488
489
490
491

def get_instance_slug(self) -> str:
    """Return the configured instance ``owner/name`` without contacting LaminHub."""
    config = self._get_config()
    return f"{config.owner}/{config.name}"

get_api_key() -> str | None

493
494
495

def get_api_key(self) -> str | None:
    """Return the connection's Lamin API key, or ``None`` for anonymous access."""
    return self._get_config().api_key

get_cache_dir() -> str | None

497
498
499

def get_cache_dir(self) -> str | None:
    """Return the connection's ``cache_dir`` extra: the lamindb cache shared by virtualenv tasks."""
    return self._get_config().cache_dir

get_instance() -> LaminDBInstance

501
502
503
504
505
506
507
508
509
510
511
512

def get_instance(self) -> LaminDBInstance:
    """
    Return the configured instance (resolves its id and regional API URL on first use).

    Raises [`LaminDBVersionError`][airflow.providers.lamindb.exceptions.LaminDBVersionError] if the
    instance runs lamindb older than 2.8.1.
    """
    if self._instance is None:
        config = self._get_config()
        settings = self._send("GET", self._settings_url(config))
        self._instance = self._instance_from_settings(config, settings)
    return self._instance

get_schema() -> dict[str, Any]

514
515
516
517
518
519

def get_schema(self) -> dict[str, Any]:
    """Return the instance schema: ``{module: {model: {"table_name": ..., "fields": ...}}}``."""
    if self._schema is None:
        instance = self.get_instance()
        self._schema = self._send("GET", f"{instance.api_url}/instances/{instance.id}/schema")
    return self._schema

get_registry(registry: str | Registry) -> Registry

521
522
523
524
525
526
527

def get_registry(self, registry: str | Registry) -> Registry:
    """Resolve a registry name such as ``core.artifact`` or ``bionty.gene``."""
    if isinstance(registry, Registry):
        return registry
    if registry.lower().startswith("hubmodule."):
        return resolve_registry({}, registry)
    return resolve_registry(self.get_schema(), registry)

query_records(
    registry: str | Registry,
    filter: FilterLike | None = None,
    *,
    select: Sequence[str] | None = None,
    order_by: Sequence[str | Mapping[str, Any]] | None = None,
    limit: int = 50,
    offset: int = 0,
    include_foreign_keys: bool = True,
    search: str | None = None,
    branch_ids: Sequence[int] | None = None,
) -> list[dict[str, Any]]

529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579

def query_records(
    self,
    registry: str | Registry,
    filter: FilterLike | None = None,
    *,
    select: Sequence[str] | None = None,
    order_by: Sequence[str | Mapping[str, Any]] | None = None,
    limit: int = 50,
    offset: int = 0,
    include_foreign_keys: bool = True,
    search: str | None = None,
    branch_ids: Sequence[int] | None = None,
) -> list[dict[str, Any]]:
    """
    Query records of a registry.

    :param registry: Registry name such as ``core.artifact``, ``core.run`` or ``bionty.gene``,
        or a [`LaminDBRegistry`][airflow.providers.lamindb.utils.enums.LaminDBRegistry] member.
    :param filter: Filter built with [`F`][airflow.providers.lamindb.utils.filters.F], e.g.
        ``F(ArtifactField.SUFFIX) == ".csv"``, or a LaminHub REST filter such as
        ``{"suffix": {"eq": ".csv"}}``, see https://docs.lamin.ai/rest
    :param select: Fields (and relations such as ``created_by(handle)``) to return.
    :param order_by: Sort order, for example ``["-created_at"]``. Defaults to ascending ``id``.
    :param limit: Maximum number of records; requests are paginated in pages of 200.
    :param offset: Number of records to skip.
    :param include_foreign_keys: Include foreign key columns (``run_id``, ``branch_id``, ...).
    :param search: Full-text search string.
    :param branch_ids: Restrict the query scope to these branch ids.
    """
    reg = self.get_registry(registry)
    instance = self.get_instance()
    results: list[dict[str, Any]] = []
    while len(results) < limit:
        page_size = min(MAX_PAGE_SIZE, limit - len(results))
        url, params, body = self._records_request(
            instance,
            reg,
            filter=filter,
            select=select,
            order_by=order_by,
            limit=page_size,
            offset=offset + len(results),
            include_foreign_keys=include_foreign_keys,
            search=search,
            branch_ids=branch_ids,
        )
        page = self._check_page(self._send("POST", url, params=params, json=body), reg)
        results.extend(page)
        if len(page) < page_size:
            break
    return results

get_records_by_ids(
    registry: str | Registry, ids: Iterable[int], filter: FilterLike | None = None
) -> dict[int, dict[str, Any]]

581
582
583
584
585
586
587
588
589
590

def get_records_by_ids(
    self, registry: str | Registry, ids: Iterable[int], filter: FilterLike | None = None
) -> dict[int, dict[str, Any]]:
    """Fetch records by id, optionally restricted by an additional filter. Returns ``{id: record}``."""
    records: dict[int, dict[str, Any]] = {}
    for chunk in chunk_ids(ids):
        combined = combine_filters({"id": {"in": chunk}}, filter)
        for record in self.query_records(registry, combined, limit=len(chunk)):
            records[record["id"]] = record
    return records

query_dbwrites(
    filter: FilterLike | None = None,
    *,
    after_id: int | None = None,
    limit: int = MAX_PAGE_SIZE,
    descending: bool = False,
) -> list[dict[str, Any]]

592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613

def query_dbwrites(
    self,
    filter: FilterLike | None = None,
    *,
    after_id: int | None = None,
    limit: int = MAX_PAGE_SIZE,
    descending: bool = False,
) -> list[dict[str, Any]]:
    """
    Query the LaminHub database write log, ordered by id.

    Every ``INSERT``, ``UPDATE`` and ``DELETE`` on the instance database is logged with
    ``table_name``, ``sqlrecord_id``, ``event_type``, ``branch_id`` and ``data``. For updates,
    ``data`` holds the *previous* values of the changed fields; for deletes the full previous
    row; for inserts it is ``null``.
    """
    return self.query_records(
        DBWRITE_REGISTRY,
        self._dbwrite_filter(filter, after_id),
        order_by=[{"field": "id", "descending": descending}],
        limit=limit,
    )

get_latest_dbwrite_id() -> int

615
616
617
618

def get_latest_dbwrite_id(self) -> int:
    """Return the id of the most recent database write (``0`` if there is none)."""
    rows = self.query_dbwrites(limit=1, descending=True)
    return int(rows[0]["id"]) if rows else 0