Skip to main content

Ingestion Service

Client helpers live in kamiwaza_sdk/services/ingestion.py and wrap the /ingestion/ingest API surface exposed by Kamiwaza. DDE connector triggering uses the separate /dde/connectors route.

Usage​

from kamiwaza_sdk import KamiwazaClient

client = KamiwazaClient("https://localhost/api", api_key="...")  # or set authenticator

# Active ingestion (runs immediately)
resp = client.ingestion.run_active(
    "s3",
    bucket="kamiwaza-sdk-tests",
    prefix="sdk-integration",
    endpoint_url="http://localhost:19100",
    region="us-east-1",
    aws_access_key_id="minioadmin",
    aws_secret_access_key="minioadmin",
)
print(resp.urns)

# Emit a Metadata Change Proposal
client.ingestion.emit_mcp({"entityType": "dataset", "changeType": "UPSERT", "entityUrn": resp.urns[0]})

# Register a job for polling (registration does not execute the cron schedule)
from kamiwaza_sdk.schemas.ingestion import IngestJobCreate

job = IngestJobCreate(
    job_id="nightly-s3-sync",
    schedule="0 3 * * *",
    source_type="s3",
    conn_args={"bucket": "...", "recursive": True},
)
client.ingestion.schedule_job(job)
status = client.ingestion.get_job_status(job.job_id)
print(status.status, status.last_run)

# Trigger an approved DDE data-source connector (requires manage permission)
trigger = client.ingestion.trigger_dde_connector_ingest("9e7fc7b1-e817-47a3-b360-872f09f379bb")
print(trigger.status)

Routing note: the FastAPI router is included under /ingestion, so the fully-qualified paths are /ingestion/ingest/run, /ingestion/ingest/jobs, etc.

Live capability verification​

test_run_now_owner_job_and_dde_trigger requires a private JSON file named by KAMIWAZA_DDE_TRIGGER_FIXTURE. It must contain allow_repeat_trigger: true, the ID of a dedicated disposable DDE connector, its expected_source_ref seed document, and a dedicated non_manager_api_key. The non-manager must have read access to that same connector but not manage access. The requester needs platform-admin access to read back the dataset owner grant. Both clients require verified TLS; configure a trusted CA rather than disabling certificate checks. Keep the fixture file outside the repository. The test triggers ingestion and leaves the platform's job and document history in place; do not point it at a shared production connector. Without the approved fixture, the complete-claim test skips rather than emitting a passing capability record.

Register source metadata​

client.ingestion.register_source registers an S3 bucket/prefix or a PostgreSQL database/schema in the selected Workroom. It creates source metadata without enumerating child datasets or materializing source data. It requires a non-Global Workroom and current membership; ReBAC-enabled deployments also enforce ingestion authority; an explicit connection workroom_id must match the selected scope.

from kamiwaza_sdk.schemas.registered_sources import S3SourceConnection, S3SourceRegistration

with client.workroom_scope(workroom_id) as scoped:
    source = scoped.ingestion.register_source(S3SourceRegistration(
        connection=S3SourceConnection(bucket="team-data", prefix="reports", secret_name=secret_urn)
    ))
    page = scoped.catalog.containers.list_registered_sources(page_size=50)
    # Registration alone returns no child datasets. Scan the registered target explicitly.
    scan = scoped.ingestion.run_active("s3", container_urn=source.container_urn)

For PostgreSQL use PostgresSourceRegistration(connection=PostgresSourceConnection(...)). Supply a password-free uri, or host, port, database and user; the existing dbname/username aliases remain available. schema_name serializes as schema. A new PostgreSQL source needs password_secret_name (or secret_name) referencing a Catalog secret in the same Workroom. Re-registration may reuse the source's existing reference.

Requests accept Catalog credential references rather than inline passwords or S3 keys. The server validates location identity and secret scope. S3 endpoint URLs cannot contain credentials, queries or fragments; PostgreSQL URIs cannot contain passwords, queries or fragments. Do not log credentials or credential-bearing request objects.

list_registered_sources(cursor=..., page_size=...) returns one typed page. Continue with next_cursor until it is absent; cursors are opaque. The summary omits child dataset enumeration and credential-reference properties. Use catalog.containers.get to inspect a selected target after scanning.

run_active performs the explicit scan and returns created dataset URNs. Registration and scans are separate from data materialization and retrieval. Deleting source metadata does not mean source bytes or database rows were deleted; account for referenced datasets and credentials before removing their containers.

Registered PostgreSQL source containers currently reject ordinary container deletion. Retiring an entire disposable Workroom removes its source metadata; this is broader than unregistering one source. Core currently has no public PostgreSQL source-unregister route.

The owned registered-source integration journey requires ReBAC enabled for its administrator/viewer refusal control. ReBAC-off authorization and non-admin ingestion-group membership are separate unverified postures.