Fetching event topics and types

1. Authenticate to the cloud platform via the Python shell.

   The following variables should be available now:

   >>> base_url  # the base URL of the API
   '<the Acronis data center URL>/api/event_manager/v1'
   >>> auth  # the 'Authorization' header value with the access token
   {'Authorization': 'Bearer eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiIsImtpZCI6ImMwMD...'}
   >>> tenant_id # the ID of the partner tenant that can be accessed with the token
   'ede9f834-70b3-476c-83d9-736f9f8c7dae'
   

2. Define a variable named topic_id, and then assign with the ID of the topic to get the event types for to this variable:

   >>> topic_id = 'cti.a.p.em.topic.v1.0~a.p.tenant.v1.0'
   

3. Send a GET request to the /topics endpoint:

   >>> response = requests.get(f'{base_url}/topics', headers=auth)
   

4. Check the status code of the response:

   >>> response.status_code
   200
   

   Status code 200 means that the request was successful.

   Also, the response body contains a list of available event topics formatted as a JSON text. When converted to an object, it will look as follows:

   >>> pprint.pprint(response.json())
   [{'archive_consolidation': 'NONE',
   'description': "Task manager's events topic",
   'id': 'cti.a.p.em.topic.v1.0~a.p.task.v1.0',
   'ordering': 'EVENTMANAGER',
   'persistency': True,
   'producer_consolidation': 'NONE',
   'retention': 'P7D'},
   {'archive_consolidation': 'TTS',
   'description': "Account Server's tenant topic",
   'id': 'cti.a.p.em.topic.v1.0~a.p.tenant.v1.0',
   'ordering': 'PRODUCER',
   'persistency': True,
   'producer_consolidation': 'TTS',
   'retention': 'P180D'}]
   

5. Check if the topic you are interested in is in the list of available topics:

   >>> topics = response.json()
   >>> any(topic_id == topic['id'] for topic in topics)
   >>> True
   

   If topic is not in the list, you may need to check the permissions or the topic ID you are using.

6. Define a variable named params, and then assign with the ID of the topic which event types to get to this variable:

   >>> params = {
   ...     "topic_id": topic_id
   ... }
   

7. Send a GET request to the /types endpoint:

   >>> response = requests.get(f'{base_url}/types', headers=auth, params=params)
   

8. Check the status code of the response:

   >>> response.status_code
   200
   

   Status code 200 means that the request was successful.

   Also, the response body contains a list of event types for the request topic formatted as a JSON text. When converted to an object, it will look as follows:

   >>> pprint.pprint(response.json())
   [{'description': 'Tenant barrier attribute updated event',
   'id': 'cti.a.p.em.event.v1.0~a.p.tenant.barrier_updated.v1.0',
   'topic_id': 'cti.a.p.em.topic.v1.0~a.p.tenant.v1.0'},
   {'description': 'Tenant created event',
   'id': 'cti.a.p.em.event.v1.0~a.p.tenant.created.v1.0',
   'topic_id': 'cti.a.p.em.topic.v1.0~a.p.tenant.v1.0'}]
   

9. Collect the list of event types that you would like to subscribe to and listen for and store them in the event_types variable:

   >>> events = response.json()
   # A set of event types to listen for
   >>> requested_events = {
   ...     'cti.a.p.em.event.v1.0~a.p.tenant.enabled.v1.0',
   ...     'cti.a.p.em.event.v1.0~a.p.tenant.disabled.v1.0',
   ... }
   >>> event_types = [event['id'] for event in events if event['id'] in requested_events]
   >>> event_types
   ['cti.a.p.em.event.v1.0~a.p.tenant.barrier_updated.v1.0', 'cti.a.p.em.event.v1.0~a.p.tenant.created.v1.0']
   

   Note

   The list of event_types may be empty if requested events are not listed in the response.