Publish Signals

pyclarify.client.Client.publish_signals(signal_ids: List[pyclarify.fields.constraints.ConstrainedStrValue] = [], items: List[pyclarify.views.items.Item] = [], items_by_signal: Dict[pyclarify.fields.constraints.ConstrainedStrValue, pyclarify.views.items.Item] = {}, create_only: bool = False, integration: str = None) pyclarify.views.generics.Response

Publishes one or multiple signals to create one or multiple items, and creates or updates a set of signals with the provided metadata. Each signal is uniquely identified by its signal ID in combination with the integration ID. Mirroring the Clarify API call admin.publishSignals .

Parameters
  • signal_ids (List['<SIGNAL_ID>']) – List of strings to be the signal ID of the signal.

  • items (List[ Item ]) – List of Item object that contains metadata for a Item. Click here for more information.

  • items_by_signal (Dict[ResourceID, Item]) – Dictionary with IDs of signals mapped to Item metadata.

  • create_only (bool, default False) – If set to True, skip update of information for existing Items. That is, all signal_ids that map to existing items are silently ignored.

  • integration (str Default None) – Integration ID in string format. None means using the integration in credential file.

Returns

Response.result.data is a dictionary mapping <SIGNAL_ID> to <ITEM_ID>.

Return type

Response

Examples

>>> client = Client("./clarify-credentials.json")

Publishing by using a dictionary.

>>> from pyclarify import Item
>>> item = Item(
...     name = "Home temperature",
...     description = "Temperature in the bedroom",
...     labels = {
...         "data-source": ["Raspberry Pi"],
...         "location": ["Home"]
...     },
...     visible=True
... )
>>> items_dict = {
...     "<SIGNAL_ID>": item
... }
>>> response = client.publish_signals(
...     items_by_signal=item_dict
... )

Publishing using arrays.

>>> from pyclarify import Item
>>> item = Item(
...     name = "Home temperature",
...     description = "Temperature in the bedroom",
...     labels = {
...         "data-source": ["Raspberry Pi"],
...         "location": ["Home"]},
...     visible=True
... )
>>> client.publish_signals(signal_ids=['SIGNAL_ID'], items=[item], create_only=False)
Response

In case of a valid return value, returns a pydantic model with the following format:

>>> jsonrpc = '2.0'
... id = '1'
... result = PublishSignalsResponse(
...                    itemsBySignal = {'SIGNAL_ID': SaveSummary(
...                           id='ITEM_ID',
...                           created=True,
...                           updated=False )})
... error = None

In case of the error the method return a pydantic model with the following format:

>>> jsonrpc = '2.0'
... id = '1'
... result = None
... error = Error(
...         code = '-32602',
...         message = 'Invalid params',
...         data = ErrorData(trace = <trace_id>, params = {})
... )