Example: Integration

11 min

End-to-end walkthrough: configure a cloud connector instance, add topics that map device tags to remote names, enable streaming.

Setup

from litmussdk.integrations.common import _integrations, _topics

1. List Available Providers

providers = _integrations.cloud_providers()
rest_provider = next(p for p in providers if p["Name"] == "REST")
provider_id = rest_provider["ID"]

# What does this provider need for instance config?
schema = _integrations.cloud_config_schema(provider_id)

# What does each topic need?
tag_schema = _integrations.cloud_config_tag_schema(provider_id)

2. Create an Instance

instance = _integrations.cloud_connector_create(
    provider_id=provider_id,
    provider_config={
        "name": "Plant-Floor-REST",
        "url": "https://my-cloud.example.com/ingest",
        "auth": {"type": "bearer", "token": "..."},
    },
)
instance_id = instance["InstanceID"]

3. Add Topics (Outbound)

Each topic maps a local Litmus tag/topic to a remote name in the cloud system.

topic_a = _topics.create_topic(
    integration_instance_id=instance_id,
    is_local_to_remote=True,                # outbound
    local_name="plant/pump/01/pressure",
    remote_name="pump-pressure",
    enabled=True,
    description="Pump pressure",
    config={},
)

topic_b = _topics.create_topic(
    integration_instance_id=instance_id,
    is_local_to_remote=True,
    local_name="plant/pump/01/flow_rate",
    remote_name="pump-flow",
    enabled=True,
    description="Pump flow rate",
    config={},
)

4. Enable the Instance

_integrations.cloud_connector_enable_disable(instance_id, enable=True)

Topics defined with enabled=True start streaming immediately.

5. Toggle Topics in Bulk

all_topics = _topics.list_topics(instance_id)
all_topic_ids = [t["TopicID"] for t in all_topics]

# Disable all temporarily (e.g. during a maintenance window)
_topics.enable_disable_multiple_topics(instance_id, all_topic_ids, enable=False)

# Re-enable
_topics.enable_disable_multiple_topics(instance_id, all_topic_ids, enable=True)

6. Import Topics from DeviceHub

Bulk-import all tags on a device into this instance as topics:

_topics.import_from_dh(
    integration_instance_id=instance_id,
    device_id=device_id,
    wildcard=False,   # or True to also include sub-topics
)

7. CSV Export / Import

csv_text = _topics.download_csv(instance_id)
with open("topics.csv", "w") as f:
    f.write(csv_text)

# Apply on another Edge:
_topics.upload_csv(instance_id, file_path="topics.csv")

8. GraphQL Subscription API (4.0.x only)

The same flow is also available via GraphQL with finer control. This is the recommended path on 4.0.x:

from litmussdk.integrations.common import _gql_streaming

# Bulk import from multiple sources
_gql_streaming.import_topics(
    instance_id=instance_id,
    device_hub=[{"deviceId": device_id, "tagIds": [tag_id]}],
    digital_twins=[{"instanceId": dt_id}],
    analytics=[{"processorId": proc_id}],
)

9. Cleanup

_topics.delete_multiple_topics(instance_id, all_topic_ids)
_integrations.cloud_connector_instance_delete(instance_id)