Publications API

Subscribe to change events from DealCloud via the Publications API. The SDK exposes topic discovery, long-poll (poll_events / stream_poll_events), acknowledgement, bootstrap, and blocking subscribe / subscribe_async helpers. HTTP uses intapp-rest-client.

Overview

Typical flow:

1. get_topics() — list topic names (for example data, schema, user).
2. poll_events() — long-poll for batches of PublicationEventVm objects (eventId, topicName, topicOffset, entityType, payload, …).
3. acknowledge_topic_offsets() — after successful processing, acknowledge the highest offset per topic (use topic_offsets_from_events() to build the payload).
4. Optionally init_bootstrap() before draining a large initial backlog; use a low count during bootstrap so each response stays bounded.

get_topics()

from dealcloud_sdk import DealCloud, DealCloudConfig
 
config = DealCloudConfig(
    siteUrl="yoursite.dealcloud.com",
    clientId=12345,
    clientSecret="your-secret",
)
dc = DealCloud.from_config_object(config)
 
topics = dc.get_topics()
# e.g. ["data", "schema", "user"]
for name in topics:
    print(name)

poll_events()

Server long-poll wait is time_out_ms (milliseconds). count caps how many events are returned per request.

events = dc.poll_events(
    topics=["data"],
    count=50,
    time_out_ms=30_000,
)
 
for e in events:
    print(e.get("topicName"), e.get("topicOffset"), e.get("entityType"))

Acknowledging events

if events:
    offsets = dc.topic_offsets_from_events(events)
    dc.acknowledge_topic_offsets(offsets)

subscribe() and subscribe_async()

Blocking loop: polls, invokes callback per event, then optionally acknowledges.

def on_event(event: dict) -> None:
    print(event.get("topicName"), event.get("entityType"))
 
dc.subscribe(
    topics=["data"],
    callback=on_event,
    time_out_ms=30_000,
    poll_interval=5,
    count=100,
    auto_acknowledge=True,
)

Async variant:

import asyncio
 
async def on_event_async(event: dict) -> None:
    await asyncio.sleep(0)  # replace with your I/O
    print(event.get("topicName"))
 
async def main():
    await dc.subscribe_async(
        topics=["data"],
        callback=on_event_async,
        time_out_ms=30_000,
    )
 
# asyncio.run(main())

init_bootstrap()

# entity_type / entity_format are API enum integers — see OpenAPI / your site docs
dc.init_bootstrap(entity_type=0, entity_format=0)

Filtered handler pattern

def filtered_handler(event, object_filter=None):
    payload = event.get("payload") or {}
    if object_filter is not None:
        # Shape depends on topic and site; inspect payload keys for your integration
        if payload.get("objectId") != object_filter:
            return
    process_event(event)
 
def process_event(event):
    ...

Best practices

1. Idempotent handlers — events can be redelivered; safe replays should not corrupt downstream state.
2. Ack only after success — if subscribe / subscribe_async raises in the callback, the batch is not acknowledged (may be redelivered).
3. Bootstrap — use init_bootstrap and small count when draining large data backlogs.
4. Large responses — use stream_poll_events and stream-parse JSON if payloads exceed comfortable memory limits.

Related

• Delta Sync — polling-based change detection via entry history
• History API — historical change data
• API reference — full Publications method list