Python Microsoft Fabric - Reading recent events
Query the last few minutes of events in an eventhouse, on demand from a service or every minute on a schedule.
This page queries the events of the last 15 minutes in an eventhouse, from a service on demand and from the scheduler every minute. An eventhouse is the place Fabric keeps events so they can be queried, and every event an eventstream receives can be stored in one - here the eventhouse Operations Events.
This is polling. When a service must run the moment an event happens, the receiving events page is the one to use.
The question in KQL
KQL is the query language of eventhouses. The query below counts the events of one type from the last 15 minutes, per location, and this is what the Fabric KQL editor shows for it:
Events
| where event_type == 'appointment_cancelled'
| where occurred_at > ago(15m)
| summarize cancelled = count() by location
The connection
The eventhouse answers queries over HTTPS at /v1/rest/query on its Query URI, and the token it takes comes from a Bearer token definition. Under Security → Bearer tokens:
| Field | Value |
|---|---|
| Name | Fabric Eventhouse Token |
| Username | The application (client) ID of your app registration |
| Password | Its client secret |
| Auth endpoint | https://login.microsoftonline.com/<tenant ID>/oauth2/v2.0/token |
| Scopes | https://kusto.kusto.windows.net/.default |
Then, under Connections → REST → Outgoing connections:
| Field | Value |
|---|---|
| Name | Operations Events |
| Host | The Query URI, e.g. https://<cluster>.kusto.fabric.microsoft.com |
| URL path | /v1/rest/query |
| Data format | JSON |
| Security | Fabric Eventhouse Token |
In enmasse YAML:
security:
- name: Fabric Eventhouse Token
type: bearer_token
username: <application (client) ID>
password: Zato_Enmasse_Env.FabricEventhouseSecret
auth_endpoint: https://login.microsoftonline.com/<tenant ID>/oauth2/v2.0/token
scopes: https://kusto.kusto.windows.net/.default
outgoing_rest:
- name: Operations Events
host: https://<cluster>.kusto.fabric.microsoft.com
url_path: /v1/rest/query
data_format: json
security: Fabric Eventhouse Token
The service
The request body names the database and carries the query. The reply is a list of tables, the first of which holds the result as a list of column descriptions and a list of rows, and the service zips the two into dicts.
# -*- coding: utf-8 -*-
# Zato
from zato.common.typing_ import anydict, dictlist
from zato.server.service import Service
def to_rows(reply:'anydict') -> 'dictlist':
""" Turns the eventhouse's reply into a list of dicts keyed by column name.
"""
# The result is the first table of the reply ..
tables = reply['Tables']
result = tables[0]
# .. its column names are in one list ..
column_names = []
for column in result['Columns']:
column_names.append(column['ColumnName'])
# .. and the rows in another.
out = []
for values in result['Rows']:
pairs = zip(column_names, values)
row = dict(pairs)
out.append(row)
return out
class RecentCancellations(Service):
name = 'events.recent-cancellations'
def handle(self):
# The query and the database it runs against ..
query = """
Events
| where event_type == 'appointment_cancelled'
| where occurred_at > ago(15m)
| summarize cancelled = count() by location
"""
request = {
'db': 'Operations Events',
'csl': query,
}
# .. posted to the eventhouse ..
conn = self.rest['Operations Events']
response = conn.post(self.cid, request)
# .. and its reply turned into rows.
rows = to_rows(response.data)
self.response.payload = {'rows': rows}
Invoked on demand, the response has the same rows the KQL editor showed:
Every minute
To run the same query every minute, open the Operations Events connection in the Dashboard and fill in the Scheduler, Request and Callback tabs:
| Tab | Field | Value |
|---|---|---|
| Scheduler | Run every | 1 minutes |
| Scheduler | Start time | Now |
| Request | Method | POST |
| Request | Body | {"db": "Operations Events", "csl": "Events \| where event_type == 'appointment_cancelled' \| where occurred_at > ago(15m) \| summarize cancelled = count() by location"} |
| Callback | Deliver to | Service |
| Callback | Service | events.on-recent-cancellations |
Every minute, the body is posted and the reply arrives in the callback service as its request:
# -*- coding: utf-8 -*-
# Zato
from zato.server.service import Service
class OnRecentCancellations(Service):
name = 'events.on-recent-cancellations'
def handle(self):
# The eventhouse's reply, delivered by the connection.
rows = to_rows(self.request.payload)
for row in rows:
cancelled = row['cancelled']
location = row['location']
self.logger.info(f'Cancelled in the last 15 minutes -> {cancelled} at {location}')
to_rows is the same function as above.
See also
| Page | What it covers |
|---|---|
| Receiving events as they happen | When every minute is not soon enough |
| Sending events as they happen | How the events got into the eventhouse in the first place |
| Events | The KQL request and response shape in detail |