Python Microsoft Fabric - Sending events as they happen

The Kafka connection

The bootstrap server from the Keys tab is the address, the topic name is the topic, TLS is on. Authentication is a security definition plus a mechanism, the same way every Kafka connection in Zato is set up.

With OAuth 2.0, the connection signs in as the same app registration your Fabric connection uses. Create a Bearer token definition first, under Security → Bearer tokens:

The namespace is the first part of the bootstrap server, up to .servicebus.windows.net. Then, under Connections → Message queues → Kafka → Outgoing connections, create the connection:

When the admin hands out a connection string instead of enabling sign-in for the app, the definition is a Basic Auth one with the username $$ConnectionString - the doubled $ is how a literal $ is entered, a single one would name an environment variable - and the connection string as the password, and the mechanism is PLAIN. The rest of the connection stays the same.

The same two connections in enmasse YAML, the OAuth 2.0 one first:

security:
  - name: Fabric Events Token
    type: bearer_token
    username: <application (client) ID>
    password: Zato_Enmasse_Env.FabricEventsSecret
    auth_endpoint: https://login.microsoftonline.com/<tenant ID>/oauth2/v2.0/token
    scopes: https://<namespace>.servicebus.windows.net/.default

outgoing_kafka:
  - name: Fabric Events
    address: <namespace>.servicebus.windows.net:9093
    topic: <topic name>
    security: Fabric Events Token
    sasl_mechanism: OAUTHBEARER
    ssl: true

security:
  - name: Fabric Events Key
    type: basic_auth
    username: $$ConnectionString
    password: Zato_Enmasse_Env.FabricEventsConnectionString
    realm: zato

outgoing_kafka:
  - name: Fabric Events
    address: <namespace>.servicebus.windows.net:9093
    topic: <topic name>
    security: Fabric Events Key
    sasl_mechanism: PLAIN
    ssl: true

The service

The event is a dict, send turns it into JSON and hands it to the eventstream:

# -*- coding: utf-8 -*-

# Zato
from zato.server.service import Service

class PassAdmission(Service):

    input = 'admission_id', 'location', 'admitted_at'

    def handle(self):

        admission = self.request.input

        # The event the eventstream receives ..
        event = {
            'event_type': 'admission',
            'location': admission.location,
            'occurred_at': admission.admitted_at,
            'admission_id': admission.admission_id,
        }

        # .. sent to it.
        self.out.kafka['Fabric Events'].send(event)

The REST connection

The eventstream's endpoint accepts events posted over HTTPS too, to https://<namespace>.servicebus.windows.net/<topic name>/messages. Create a Bearer token definition under Security → Bearer tokens:

Then, under Connections → REST → Outgoing connections:

In enmasse YAML:

security:
  - name: Fabric Events REST Token
    type: bearer_token
    username: <application (client) ID>
    password: Zato_Enmasse_Env.FabricEventsSecret
    auth_endpoint: https://login.microsoftonline.com/<tenant ID>/oauth2/v2.0/token
    scopes: https://eventhubs.azure.net/.default

outgoing_rest:
  - name: Fabric Events REST
    host: https://<namespace>.servicebus.windows.net
    url_path: /<topic name>/messages
    data_format: json
    security: Fabric Events REST Token

The service

The endpoint takes a list of events, each wrapped in a Body field, with a content type of its own:

# -*- coding: utf-8 -*-

# stdlib
import json

# Zato
from zato.server.service import Service

class PassAdmission(Service):

    input = 'admission_id', 'location', 'admitted_at'

    def handle(self):

        admission = self.request.input

        # The event the eventstream receives ..
        event = {
            'event_type': 'admission',
            'location': admission.location,
            'occurred_at': admission.admitted_at,
            'admission_id': admission.admission_id,
        }

        body = json.dumps(event)
        payload = [{'Body': body}]
        headers = {'Content-Type': 'application/vnd.microsoft.servicebus.json'}

        # .. sent to it.
        conn = self.rest['Fabric Events REST']
        conn.post(self.cid, payload, headers=headers)