Import files from a remote connector¶

A connector is remote object storage your organization already uses — a team S3 bucket — registered with Nexus once by an administrator. Importing from one gets a dataset into a collection without downloading it to your laptop first, and without your notebook ever handling cloud credentials.

The calls are the ones you already know from Bring simulation files into a collection: collection.plan(datasource) to validate, plan.apply() to import. Only the datasource changes. What is different is where the file list comes from — Nexus lists the remote prefix server-side, seals the result, and validates that sealed list against your collection schema before a single byte moves.

Before you start¶

Someone with connector administration rights registers the connector, and hands you the first three values below. The other two are a project and collection that already exist — Bring simulation files into a collection makes those.

The runnable scripts read all five from those variables, and authenticate the same way every other example does — see Authentication for MIURA_NEXUS_API_KEY and the deployment URL.

Connector import is async only. Inventory and transfer both poll a remote job, so collection.plan() raises TypeError for a ConnectorDataSource on the sync client (from miura import Nexus), and the connector administration and inventory namespaces are async too. Every snippet on this page therefore uses AsyncNexus — that is the one place where a connector import differs from the local upload it otherwise mirrors.

Snippets continue one session. The first opens it; the rest run inside that async with block, against a collection you fetched from your project:

from miura.api import AsyncNexus

async with AsyncNexus() as nexus:
    project = await nexus.get_project(project_name)
    collection = await project.get_collection(collection_name)

How the flow fits together¶

Two services split the work, and the split explains most of the API's shape: the data source service owns what a connector is, the ingestion service owns what you are about to import.

Importing never puts credentials in your process. They stay with the platform, which materializes them inside the workers that do the listing and copying — nothing you pass to ConnectorDataSource is secret. Registering a connector is the one place you handle them, and get_credentials() hands them back; both are administration, not import.

Find and test a connector¶

Start from the types your deployment supports — a type declares which capabilities a connector of that type can be asked to prove.

from miura.api import AsyncNexus

async with AsyncNexus() as nexus:
    for connector_type in await nexus.connectors.list_types():
        capabilities = ",".join(connector_type["capabilities"])
        print(f"{connector_type['key']:<16} {connector_type['display_name']:<44} {capabilities}")

rclone_http      HTTP(S) endpoint (rclone)                    test_connection,list_objects,read_files
rclone_ovh_s3    OVHcloud Object Storage (rclone S3)          test_connection,list_objects,read_files,write_files
rclone_s3        S3-compatible object storage (rclone)        test_connection,list_objects,read_files,write_files

Then probe the connector you were given. test_and_wait submits a test, polls until it reaches a terminal state, and raises on anything that is not green — it fails closed rather than letting a broken connector reach an inventory job.

status = await nexus.connectors.test_and_wait(
    connector_id,
    {"requested_capabilities": ["test_connection", "list_objects"]},
)
print(status["status"], status["pinned_revision"], status["capabilities_verified"])

success 2 ['test_connection', 'list_objects']

Ask for list_objects when you are about to inventory: test_connection reaches the connector's root, which proves the endpoint answers and the credentials are accepted, but not that they can walk the prefix you are about to import. requested_capabilities is required — the service rejects an empty test body — so the SDK defaults it to ["test_connection"] when you pass nothing. It is recorded on the test run and comes back on the response beside the result.

capabilities_verified reports what the probe established, never what the connector's type declares. A capability that fails takes the whole test down with it, so on a green result the list is exactly what you asked for — and on a red one test_and_wait raises rather than hand you a partial pass.

The test pins a connector revision. Every edit to a connector bumps that revision, and the revision recorded at inventory time is checked again at import time. If the connector is rotated in between, the import is refused with PLAN_SOURCE_EVIDENCE_MISMATCH rather than transferring against configuration nobody validated.

Register a connector¶

Registration is a single call — there is no separate enrollment step. config describes where the storage is; secrets is the only place credentials appear.

created = await nexus.connectors.create_connector(
    {
        "name": "team-openfoam-bucket",
        "type_key": "rclone_s3",
        "driver_kind": "rclone",
        "config": {
            "provider": "Other",
            "region": "gra",
            "bucket": "customer-bucket",
            "endpoint": "https://s3.gra.io.cloud.ovh.net",
            "prefix": "for_dev",
        },
        "secrets": {
            "access_key_id": access_key_id,
            "secret_access_key": secret_access_key,
        },
    }
)
print(created["connector_id"], created["revision"], created["secret_ref"])

conn_5329aaa438a94473 1 openbao://kv/nexus/data-source-service/connectors/<org>/conn_5329aaa438a94473/<ref>

The secret does not come back from create_connector. What you get is a secret_ref scoped to your organization and that connector, which the inventory and transfer workers re-validate before they are allowed to read it.

update_connector(connector_id, payload) patches a registration; delete_connector(connector_id) removes it. Each connector type takes a different config shape — read it from get_type(type_key) rather than guessing.

A patch bumps the revision when it changes something an import depends on — config, secrets, driver_kind, type_key. Renaming a connector does not: the revision tracks what the workers materialize, and a display name is not part of that.

Choose what to import¶

ConnectorInventoryScope is the question "what does everything mean here". It is answered once, server-side, and the answer is frozen.

from miura.sdk import ConnectorDataSource, ConnectorInventoryScope

datasource = ConnectorDataSource(
    connector_id="conn_5329aaa438a94473",
    inventory_scope=ConnectorInventoryScope(
        source_prefix="s3://customer-bucket/for_dev",
        include_glob="**/*.h5",
    ),
    transfer_binding_id=transfer_binding_id,
)

Two properties of that scan are worth knowing, because they are what makes the later validation report trustworthy:

Sealed. The platform re-reads the listing the worker produced and checks it independently — schema, path safety, no duplicates, exact object count, the manifest's own byte count and SHA-256 — before marking the job successful. The worker's claim that it finished is not taken as proof. The scan also totals the sizes of the objects it listed; that total comes back as object_bytes on the manifest, and it is what the import reports as its size. Do not confuse it with the byte count the sealing check verifies — that one measures the manifest document, a few kilobytes for a scan of 27 objects whatever those objects weigh, and it stays server-side.

Bounded, never truncated. A prefix with more objects than max_keys fails the inventory with INVENTORY_LIMIT_EXCEEDED. It does not silently import the first 20 000 files. Narrow include_glob or import a deeper prefix instead.

Running the same scope again within the reuse window returns an alias of the earlier scan rather than re-listing — same sealed manifest, new auditable job row, reused: true and reused_from_job_id pointing at the scan it aliases. Changing the connector's revision invalidates the reuse, and force_refresh=True bypasses it.

Those two fields are on the job, not the manifest, so they come back from nexus.connector_inventory.start() and get() — the raw job dicts. resolve(), and plan() through it, hand you the sealed manifest and the job binding without them; reach for start() when whether a scan was reused is the thing you are trying to find out. connector_inventory.py Steps 4-5 show both halves.

Review before anything moves¶

collection.plan() takes either datasource and returns the same handle. For a connector it runs the inventory scan, seals the manifest, and validates it against the collection schema. No bytes move.

from miura.display import print_ingestion_plan
from miura.sdk import UploadMode

plan = await collection.plan(datasource, mode=UploadMode.INSERT)
print_ingestion_plan(plan)

Ingestion plan
──────────────────────────────────────────────────
   Plan ID:  igplan_4f2…4c3a8f16 igplan_4f2c1a9db7e04c3a8f16
   Mode:     INSERT
   Type:     CONNECTOR_TRANSFER
   Gate:     passed
   State:    SUCCEEDED
   Admissible: True
   Candidates: 27

The plan id appears twice on purpose: an abbreviated form to read, then the full one to copy. Every output on this page comes from the same run against the same 27-object prefix, so the identifiers and counts line up across all four.

plan.print_validation_report() gives the record-level detail — which remote paths the schema admitted and which it rejected, and why.

report = await plan.print_validation_report(max_issues=20)
print(len(report.passed_records), "admitted;", len(report.failed_records), "rejected")

Plan validation report (failed)
──────────────────────────────────────────────────
   Plan ID: igplan_4f2…4c3a8f16 igplan_4f2c1a9db7e04c3a8f16
   Manifest: 3b9a4c17e2f8d05a…
   Passed (27):
   • /case_000001/system/controlDict
   • /case_000001/0/U
   • /case_000001/0/p
   • /case_000001/post/post000001_+0.00000e+00.h5
   • /case_000001/post/post000002_+1.00000e-01.h5
   … and 22 more

27 admitted; 0 rejected

The (failed) in that title is the report's default heading, not a verdict on your plan — read plan.is_admissible and report.failed_records for the answer. Gate the transfer on the former:

if not plan.is_admissible:
    raise SystemExit(f"{plan.blocking_failure_count} blocking failures — nothing imported")

Preview: describe cases before import¶

If each remote folder represents one simulation, pass the same case_grouping= you would use for a local folder. Connector case grouping is on by default (ADR-0021 Phase A). Opt out for a process with MIURA_CONNECTOR_CASE_GROUPING=false only when your deployment does not yet support it.

The grouping API is the same one used for local files:

from miura.nexus.case_grouping import CaseGrouping

case_grouping = CaseGrouping.root_path("/", depth=1)
plan = await collection.plan(
    datasource,
    mode=UploadMode.INSERT,
    case_grouping=case_grouping,
)

Inspect plan.validation_gates and plan.is_admissible before applying. A successful grouping exposes plan.case_grouping_result and plan.cases; a failed grouping blocks plan.apply() and names case_grouping in the error. When the client has opted out, collection.plan() raises a ValueError rather than silently ignoring the grouping request.

Import¶

plan.apply() is where bytes move, and it is the same call as a local upload because the routing it needs travels on the datasource.

from miura.display import print_upload_result

result = await plan.apply()
print_upload_result(result)

Upload result
──────────────────────────────────────────────────
   Status:    SUCCEEDED
   Run:       ingr_9c74b…f0a34d5b ingr_9c74b2e1f0a34d5b
   Plan:      igplan_4f2…4c3a8f16 igplan_4f2c1a9db7e04c3a8f16
   Uploaded:  27  Failed: 0  Size: 1.5 GiB

The size comes from the sealed manifest, not the transfer. Against a deployment whose inventory never recorded object_bytes, result.total_size_bytes is None and the Size field is left out rather than shown as 0 B.

Routing is checked when you call apply(), not when you build the datasource — so a datasource built only to inventory a prefix and read the report stays valid without it. Call datasource.require_transfer_routing() if you want that check earlier; it returns (transfer_binding_id, airflow_dag_id, asset_metadata) or raises naming what is missing. Only the binding is required. The middle value is an escape hatch for redirecting a single run onto a different transfer DAG — the ingestion service picks the DAG from its own settings, so leave it unset and expect None back.

The transfer copies exactly the admitted set from the sealed manifest. Objects added to the bucket after the inventory scan ran are not picked up — the run and the report you approved describe the same files. Re-run plan() to pick up later arrivals.

await plan.apply() resolves when the transfer reaches a terminal state, and raises if that state is not green.

Read what was imported¶

Nothing here is connector-specific — imported files are collection items like any other, so the whole of Read recipe results for analysis applies.

print(await collection.count_items(), "items")
async for item, _depth in collection.walk_items(path="/", include_dirs=False):
    print(item.item_uri, item.file_size)

27 items
/case_000001/system/controlDict 1284
/case_000001/0/U 8402
/case_000001/0/p 8380
/case_000001/post/post000001_+0.00000e+00.h5 87421184
/case_000001/post/post000002_+1.00000e-01.h5 89112576

Three cases of nine files each: a small control and field set, then six result files per case. The result files are what make up almost all of the 1.5 GiB the transfer reported.

Local and connector plans compared¶

Everything you need to decide whether to apply is on both kinds of plan:

Properties that only describe a local tree raise AttributeError on a connector plan rather than report a misleading zero: data_root, total_size_bytes, gate_status, work_summary, ignored_extras, requires_execution. Grouping properties are shared: case_grouping_spec_id, case_grouping_spec_sha256, validation_gates, case_grouping_result, and cases. Branch on plan.run_type (or hasattr) when you handle both origins. The display helpers already do — print_ingestion_plan works on either.

max_files= is refused for connectors because the file list is sealed remotely:

• max_files= — planning a subset of a sealed manifest would break the binding between the plan and the scan it was validated against. Bound the candidate set with ConnectorInventoryScope(max_keys=…) at inventory time instead.

It raises ValueError at plan() time rather than being silently ignored.

Replay a sealed manifest¶

A sealed manifest can be planned again without re-listing the remote prefix — useful when you want to retry an import against exactly the file set you reviewed.

resolved = await nexus.connector_inventory.resolve(datasource)
replay = datasource.with_manifest(resolved.manifest)

plan = await collection.plan(replay, mode=UploadMode.INSERT)

with_manifest() carries transfer_binding_id and asset_metadata across, so the rebuilt datasource can still apply. A replayed manifest must carry the inventory_job_id it came from — that binding is what the plan is admitted against, so a ConnectorDataSource built on a CandidateManifest without one raises ValueError. Manifests from nexus.connector_inventory.get_manifest() always carry it.

Note what the replayed datasource does not carry: an inventory scope, and so no source_prefix. That is deliberate — a replay lists nothing, and a datasource holds either a scope or a manifest, never both. plan() reads the canonical prefix off the inventory job the manifest is bound to, so you do not have to supply one.

You can also drive the scan yourself with nexus.connector_inventory: start(connector_id, scope), wait(job_id), get(job_id), get_manifest(job_id). That is the same work plan() does for you, exposed for when you want to inspect reuse, scope hashes, or expiry before committing to a plan — and get() is what returns them, the job row with its state, reused, scope.scope_hash and expires_at. get_manifest() returns the sealed artifact instead: counts, checksum, provenance.

When something fails¶

Errors carry a typed code, and it is not in the message — plan and transfer rejections raise ValidationError, whose .code is the SDK's own coarse validation_error. The platform code is in .details:

from miura.sdk.exceptions import NexusError

try:
    result = await plan.apply()
except NexusError as exc:
    print(exc.details.get("code"))    # PLAN_SOURCE_EVIDENCE_MISMATCH
    print(exc.details.get("ingestion_run_id"))

Inventory and transfer polling are the exception: those raise RuntimeError with the code inside the message text, because the failure is reported on a job row rather than an HTTP response.

The codes worth recognizing:

INGESTION_PLAN_REQUIRED is a policy, not a bug: some environments refuse connector transfers that were not planned first. Going through collection.plan() always satisfies it.

PLAN_ALREADY_CONSUMED is a rule about plans, not about apply(). Awaiting the same plan handle twice does not raise it — the run's idempotency key is fixed when the plan is created, so the second call returns the first run's result rather than starting a second transfer. You reach the code by creating a fresh run against a spent plan; from this API, just call plan() again.

One failure carries no code at all. A prefix that exists but holds nothing is a successful scan of zero objects, so plan() refuses it locally with ValueError: connector import requires a non-empty candidate manifest, naming the prefix and glob that found nothing. It is almost always a mistyped source_prefix or an include_glob that matches none of the files that are there.

Runnable scripts¶

Each link opens that script's full source. API reference: Connectors, Datasources, Ingestion.