Fabric events

This page will show you how to work with Microsoft Fabric events in Python:

• How to subscribe to events
• How to push events
• How to run queries to read events

In Fabric nomenclature, the place you can subscribe to for and push events to is called an eventstream, i.e. it's a stream of events. It uses the Kafka protocol, so you connect to it using Zato's Kafka connections.

The storage place for events is called an eventhouse, and to run queries against it, Fabric offers a language called KQL, and the Fabric connection runs such queries for you.

Collecting Fabric config

You'll need some details from Fabric to configure Zato connections:

• The bootstrap server (Fabric name for an input Kafka endpoint to connect to)
• The topic name to use
• Namespace of the Kafka endpoint

In the example eventstream screenshot below, bootstrap server is f9d600052ccf.servicebus.windows.net:9093 and the namespace part of it is f9d600052ccf. You should also copy the topic name which in this case is h5vo4d31s95bg_eh.

To query an eventhouse, you'll need its ID, which is the part of its URL after /eventhouses/, e.g. 4ad19ee4-fc68-4612-b952-c3965ab4f3b7.

Creating Zato connections

The services on this page use two Kafka connections, created in the Zato Dashboard, and the Fabric connection from the tutorial.

Every field of each connection is also in the events reference.

Sending an event

Here's an example of of near real-time communication for when something happens in one of your systems and Fabric should know about it right away.

The service below has two simple steps:

• Build the event from the input it received
• Send it to the eventstream

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

# Zato
from zato.server.service import Service

class PassAdmission(Service):

    name = 'fabric.pass-admission'
    input = 'admission_id', 'location', 'admitted_at'

    def handle(self):

        input = self.request.input

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

        # .. get a Kafka connection ..
        conn = self.out.kafka['Fabric Events']

        # .. and send the event.
        conn.send(event)

After invoking the service you'll see your events in Fabric as below:

Receiving events

Your service attached to Kafka channels pointing to Fabric eventstreams will be invoked for events as they ocurrs. Each it's invoked, it will be a given a list of events, possibly a single-element one.

The service below will:

• Go through the events in its input
• Log the fields of each - in a real service you'd process them as needed, e.g. by invoking other systems

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

# Zato
from zato.server.service import Service

class NotifyPurchasing(Service):

    name = 'fabric.notify-purchasing'

    def handle(self):

        # The eventstream sends a list of events ..
        for event in self.request.input:

            # .. read the fields of each one ..
            item_id = event['item_id']
            location = event['location']
            quantity = event['quantity']
            reorder_level = event['reorder_level']

            # .. and log them.
            self.logger.info(f'Low stock -> {item_id} at {location}')
            self.logger.info(f'Left -> {quantity}')
            self.logger.info(f'Reorder at -> {reorder_level}')

Each time Fabric sends an event, the service will logs messages such as here:

2026-09-27 15:49:31,559 - INFO - Low stock -> ITEM-001 at Oak Hill
2026-09-27 15:49:31,560 - INFO - Left -> 15
2026-09-27 15:49:31,561 - INFO - Reorder at -> 50

Reading recent events

In this case, a Fabric eventstream writes its events to the Events table of the Operations Events eventhouse, and a service queries that table with KQL.

The service below will:

• Run a KQL query against the eventhouse to count cancelled appointments per location in the last 15 minutes.
• Return the rows

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

# Zato
from zato.server.service import Service

class RecentCancellations(Service):

    name = 'fabric.recent-cancellations'

    def handle(self):

        workspace_id = '262ddd3d-3d0d-4495-b80f-6dddda1e23a1'
        eventhouse_id = '4ad19ee4-fc68-4612-b952-c3965ab4f3b7'

        # The query to run ..
        query = """
        Events
        | where event_type == 'appointment_cancelled'
        | where occurred_at > ago(15m)
        | summarize cancelled = count() by location
        """

        # .. get a Fabric connection ..
        conn = self.microsoft.fabric['Zato Fabric']

        # .. run the query, each row comes back as a dict ..
        rows = conn.query_events(workspace_id, eventhouse_id, query)

        # .. and return the rows to our caller.
        self.response.payload = rows

After invoking the service you'll see:

[{"location": "Maple Grove", "cancelled": 1}]

See also

Learn more