Fabric API - Queries

query and refresh_sql_endpoint on the SQL analytics endpoint, query_events on an eventhouse, run_spark, open_spark_session and close_spark_session on Spark.

query sends T-SQL to a lakehouse's SQL analytics endpoint and returns rows in well under a second. run_spark sends code to a Spark session, for PySpark and for statements that write tables. query_events sends KQL to an eventhouse.

query

conn.query(workspace_id, lakehouse_id, sql, params=None)

Runs a T-SQL select against the lakehouse's SQL analytics endpoint and returns the rows as a list of dicts keyed by column name.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the lakehouse is in
`lakehouse_id`	str	The lakehouse to query
`sql`	str	The statement, in T-SQL, with `:name` markers for the values `params` supplies
`params`	dict	The values for the markers, by name, or `None` when the statement has none

Returns a list of dicts, one per row. A statement that fails raises an exception with the endpoint's error message.

Values go through params. A marker is a colon followed by a name, as in :location, and every marker in the statement needs a key of that name in params.

Tables are under the default dbo schema, so a plain table name works. A computed column needs an alias, e.g. count(*) as admission_count, otherwise it comes back under an empty name.

The first query against a lakehouse logs in to the endpoint, which takes a second or two, and the login is kept open for the calls that follow.

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

# Zato
from zato.server.service import Service

class Query(Service):

    def handle(self):

        workspace_id = '262ddd3d-3d0d-4495-b80f-6dddda1e23a1'
        lakehouse_id = '73811955-064d-481a-a7e9-f9c563124f9e'

        conn = self.microsoft.fabric['Zato Fabric']

        rows = conn.query(
            workspace_id,
            lakehouse_id,
            'select location, occupied_beds from occupancy where as_of = :as_of order by location',
            {'as_of': '2026-08-31'},
        )

        self.response.payload = {'rows': rows}

{
  "rows": [
    {"location": "Maple Grove", "occupied_beds": 37},
    {"location": "Oak Hill", "occupied_beds": 60},
    {"location": "Riverside", "occupied_beds": 96}
  ]
}

refresh_sql_endpoint

conn.refresh_sql_endpoint(workspace_id, lakehouse_id, table_names=None)

Synchronizes the SQL analytics endpoint with the latest state of a lakehouse's tables and waits until it has done so. The endpoint updates its view of the tables on its own within a few minutes of a write, and this method is for when a query has to see a write right away.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the lakehouse is in
`lakehouse_id`	str	The lakehouse whose endpoint to refresh
`table_names`	list	The tables to refresh, or `None` for all of them

Returns a dict with a value list holding one entry per table, each with tableName and status. A table whose refresh failed raises an exception with the endpoint's message.

write_table calls this method for the table it wrote. Call it directly after writing to a table through Spark or OneLake.

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

# Zato
from zato.server.service import Service

class RefreshSQLEndpoint(Service):

    def handle(self):

        workspace_id = '262ddd3d-3d0d-4495-b80f-6dddda1e23a1'
        lakehouse_id = '73811955-064d-481a-a7e9-f9c563124f9e'

        conn = self.microsoft.fabric['Zato Fabric']

        result = conn.refresh_sql_endpoint(workspace_id, lakehouse_id, ['admissions_by_location'])

        self.response.payload = {'tables': result['value']}

{
  "tables": [
    {"tableName": "admissions_by_location", "status": "Success"}
  ]
}

query_events

conn.query_events(workspace_id, eventhouse_id, query, database='')

Runs a KQL query against an eventhouse and returns its rows. The connection finds the eventhouse's address and default database itself.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the eventhouse is in
`eventhouse_id`	str	The eventhouse to query
`query`	str	The KQL query
`database`	str	The KQL database to run it against, or `''` for the eventhouse's default one

Returns a list of dicts, one per row, keyed by column name. A query the eventhouse rejects raises an exception with the eventhouse's message.

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

# Zato
from zato.server.service import Service

class EventsPerLocation(Service):

    def handle(self):

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

        conn = self.microsoft.fabric['Zato Fabric']

        query = """
        Events
        | where occurred_at > ago(1h)
        | summarize events = count() by location
        """

        rows = conn.query_events(workspace_id, eventhouse_id, query)

        self.response.payload = {'rows': rows}

{
  "rows": [
    {"location": "Riverside", "events": 12},
    {"location": "Maple Grove", "events": 7}
  ]
}

run_spark

conn.run_spark(workspace_id, lakehouse_id, session_id, code, kind='pyspark')

Runs code on a Spark session and returns the statement's output once it completes.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the lakehouse is in
`lakehouse_id`	str	The lakehouse the session belongs to
`session_id`	str	The session, from `open_spark_session`
`code`	str	The code to run
`kind`	str	`pyspark`, `sql`, `spark` or `sparkr`

Returns a dict with status, execution_count and data, where data holds the result under a key named after its format, e.g. text/plain for what the code printed. Code that fails raises an exception with the error's message.

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

# Zato
from zato.server.service import Service

class RunSpark(Service):

    def handle(self):

        workspace_id = '262ddd3d-3d0d-4495-b80f-6dddda1e23a1'
        lakehouse_id = '73811955-064d-481a-a7e9-f9c563124f9e'

        conn = self.microsoft.fabric['Zato Fabric']

        code = """
frame = spark.sql('select location, count(*) as admissions from admissions group by location')
frame.write.mode('overwrite').saveAsTable('admissions_by_location')
print(frame.count())
"""

        session_id = conn.open_spark_session(workspace_id, lakehouse_id)

        try:
            output = conn.run_spark(workspace_id, lakehouse_id, session_id, code)
        finally:
            conn.close_spark_session(workspace_id, lakehouse_id, session_id)

        data = output['data']
        printed = data['text/plain']

        self.response.payload = {'status': output['status'], 'printed': printed}

open_spark_session

conn.open_spark_session(workspace_id, lakehouse_id)

Opens a Spark session on a lakehouse and waits until it is ready, which takes from a few seconds to a couple of minutes.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the lakehouse is in
`lakehouse_id`	str	The lakehouse the session belongs to

Returns the session ID as a string. A session that does not become ready within ten minutes raises an exception.

close_spark_session

conn.close_spark_session(workspace_id, lakehouse_id, session_id)

Closes a session opened with open_spark_session.

Parameter	Type	Meaning
`workspace_id`	str	The workspace the lakehouse is in
`lakehouse_id`	str	The lakehouse the session belongs to
`session_id`	str	The session, from `open_spark_session`

Spark sessions

A service opens a session, runs as many statements on it as it needs and closes it. Fabric closes a session that is left idle after a while, and run_spark raises an exception when the session is gone.

Sessions run through Fabric's Livy endpoint, which a tenant admin has to enable under Tenant settings → Developer settings in the Fabric admin portal. Without it, open_spark_session fails with an authorization error.

Page	What it covers
Fabric API - Tables	Writing the tables the queries read
Reading data from Fabric	query with params and totals for reports
Fabric events	query_events reading recent events from an eventhouse