Python Microsoft Fabric - Looking up Fabric data from your services
Keep current names from Fabric tables at hand while processing messages, refreshed on a schedule.
A service that processes messages often gets IDs in each message and needs the names that go with them. When the names are in Fabric tables, looking each one up is too slow - a lakehouse query runs on Spark and takes seconds, and a service may get a thousand messages a minute. Instead, one service reads the tables every few minutes and puts them in the cache, and every other service reads the cache. This page has both services.
The tables in Fabric
The service below copies two tables into the cache, locations and insurers, each with an ID column and a name column:
The refresh service
One service, one query per table, and each table lands in the cache as a JSON map from ID to name. It runs from the scheduler every five minutes, and the cache entries live for ten, so a refresh that fails once leaves the previous names in place rather than nothing.
# -*- coding: utf-8 -*-
# stdlib
from json import dumps
# Zato
from zato.server.service import Service
# Which table feeds which cache key, and which columns are the ID and the name
_lookups = {
'lookups:locations': ('locations', 'location_id', 'name'),
'lookups:insurers': ('insurers', 'insurer_id', 'name'),
}
_cache_ttl = 600
class RefreshLookups(Service):
name = 'lookups.refresh'
def handle(self):
workspace_id = '11111111-2222-3333-4444-555555555555'
lakehouse_id = '66666666-7777-8888-9999-000000000000'
conn = self.microsoft.fabric['My Fabric']
counts = {}
for cache_key, (table_name, id_column, name_column) in _lookups.items():
# Read the table ..
sql = f'select {id_column}, {name_column} from {table_name}'
rows = conn.query(workspace_id, lakehouse_id, sql)
# .. turn it into a map from ID to name ..
names = {}
for row in rows:
names[row[id_column]] = row[name_column]
# .. and keep it in the cache for ten minutes.
data = dumps(names)
self.cache.set(cache_key, data, ex=_cache_ttl)
counts[table_name] = len(names)
self.response.payload = counts
To schedule the service, go to Scheduler → Config in the Dashboard, click "Create a new job" and pick the interval-based type with 5 minutes and lookups.refresh as the service.
After the first run, the cache has the two entries:
The lookup in a message service
Any service that needs a name reads the cache. This one gets a message with two IDs and fills in the two names before it passes the message on:
# -*- coding: utf-8 -*-
# stdlib
from json import loads
# Zato
from zato.server.service import Service
class ProcessInvoice(Service):
name = 'invoices.process'
input = 'invoice_id', 'location_id', 'insurer_id', 'amount'
def handle(self):
invoice = self.request.input
# The names, as Fabric had them a few minutes ago.
cached_locations = self.cache.get('lookups:locations')
cached_insurers = self.cache.get('lookups:insurers')
locations = loads(cached_locations)
insurers = loads(cached_insurers)
invoice_id = invoice.invoice_id
amount = invoice.amount
location_name = locations[invoice.location_id]
insurer_name = insurers[invoice.insurer_id]
self.logger.info(f'Invoice {invoice_id} -> {location_name}, {insurer_name}, {amount}')
A message with INV-3310, LOC-01 and INS-01 logs Invoice INV-3310 -> Riverside, Cascade Health Plan, 2275.00, and no Spark session was involved in answering it.
See also
| Page | What it covers |
|---|---|
| Keeping lookup tables up to date | How the names get into Fabric in the first place |
| Building an API on top of Fabric data | A cache in front of a query that callers ask on demand |
| Queries | query and Spark sessions in detail |