Source Systems

The Source Systems resource (client.source_systems) manages the system inventory behind your catalog: warehouses, CRM platforms, lakehouses, SaaS systems, roles, and masked credentials.

Methods

list(space_slug)

List all source systems in a space.

Parameter	Type	Description
space_slug	str	Space identifier

Returns: list[SourceSystem]

systems = await client.source_systems.list("marketing-analytics")

---

get(space_slug, source_system_id)

Get a source system by ID, including nested roles and connector summaries when returned by the API.

Parameter	Type	Description
space_slug	str	Space identifier
source_system_id	UUID	Source system UUID

Returns: SourceSystem

---

resolve_space(source_system_id)

Resolve the owning space slug for a source system.

Parameter	Type	Description
source_system_id	UUID	Source system UUID

Returns: str

---

create(space_slug, *, name, platform_type, ...)

Create a source system.

Parameter	Type	Default	Description
space_slug	str	required	Target space
name	str	required	Source system display name
platform_type	str	required	Platform type, such as snowflake, postgresql, or salesforce
description_markdown	str | None	None	Source system description
documentation_url	str | None	None	Link to system documentation
runbook_url	str | None	None	Link to operational runbook
system_owner	str | None	None	Owner name
system_owner_email	str | None	None	Owner email
support_channel	str | None	None	Support channel
contact_persons	list[dict[str, str]] | None	None	Additional contacts
icon	str | None	None	Icon identifier

Returns: SourceSystem

source_system = await client.source_systems.create(
    "marketing-analytics",
    name="Production Warehouse",
    platform_type="snowflake",
    system_owner_email="data-platform@example.com",
)

---

update(space_slug, source_system_id, **fields)

Update a source system.

Parameter	Type	Description
space_slug	str	Space identifier
source_system_id	UUID	Source system UUID
**fields	Any	Fields to update

Returns: SourceSystem

---

delete(space_slug, source_system_id)

Delete a source system. Associated connectors are unlinked by the API.

Parameter	Type	Description
space_slug	str	Space identifier
source_system_id	UUID	Source system UUID

Returns: None

Roles

create_role(space_slug, source_system_id, *, name, ...)

Create a role on a source system.

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
source_system_id	UUID	required	Source system UUID
name	str	required	Role display name
description	str | None	None	Role description
role_identifier	str | None	None	External role name or identifier
access_level	str	read	Access level
workflow_id	UUID | None	None	Linked workflow
workflow_definition_id	UUID | None	None	Linked workflow definition

Returns: SourceSystemRole

role = await client.source_systems.create_role(
    "marketing-analytics",
    source_system.id,
    name="Warehouse Readers",
    role_identifier="WAREHOUSE_READERS",
)

get_role(space_slug, source_system_id, role_id)

Get role detail, including users with access.

Returns: SourceSystemRoleDetail

update_role(space_slug, source_system_id, role_id, **fields)

Update a source-system role.

Returns: SourceSystemRole

delete_role(space_slug, source_system_id, role_id)

Delete a source-system role.

Returns: None

Credentials

Credential responses are masked. Secrets can be supplied on create or update, but are never returned by the API.

list_credentials(space_slug, source_system_id)

List credentials for a source system.

Returns: list[SourceSystemCredential]

create_credential(space_slug, source_system_id, *, name, ...)

Create a credential for a source system.

Parameter	Type	Default	Description
space_slug	str	required	Space identifier
source_system_id	UUID	required	Source system UUID
name	str	required	Credential display name
description	str | None	None	Credential description
credential_type	str	database	Credential type
connection_config	dict | None	None	Non-secret connection settings
secret	str | None	None	Secret value to store
is_active	bool	True	Whether the credential is active

Returns: SourceSystemCredential

credential = await client.source_systems.create_credential(
    "marketing-analytics",
    source_system.id,
    name="Read-only service account",
    credential_type="service_account",
    connection_config={"warehouse": "ANALYTICS"},
    secret="...",
)

update_credential(space_slug, source_system_id, credential_id, **fields)

Update a credential. Pass secret= to rotate the secret.

Returns: SourceSystemCredential

delete_credential(space_slug, source_system_id, credential_id)

Delete a credential.

Returns: None

Sync Client

The synchronous client exposes the same methods without await:

from qarion import QarionSyncClient

with QarionSyncClient(api_key="qk_...") as client:
    systems = client.source_systems.list("marketing-analytics")