Python Microsoft Fabric - Tables

Loading rows and files into lakehouse tables and querying them with SQL.

Tables are where a lakehouse keeps queryable data, and this page shows how to put rows in and get them back out. You create a Fabric connection in the Dashboard and every method below is available on it.

Writing rows to a table

conn.write_table takes a list of dicts and turns it into a table - the rows travel as files through OneLake, one load operation makes them the table's data, and the call returns once the load completes.

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

# Zato
from zato.server.service import Service

class WriteDailyTotals(Service):

    input = 'workspace_id', 'lakehouse_id'

    def handle(self):

        # Get the connection by its Dashboard name
        conn = self.microsoft.fabric['My Fabric']

        rows = [
            {'region': 'EMEA', 'total': 1250.50},
            {'region': 'APAC', 'total': 875.25},
        ]

        # Write the rows, replacing what the table held before
        conn.write_table(self.request.input.workspace_id, self.request.input.lakehouse_id, 'daily_totals', rows)

        self.response.payload = {'status': 'written', 'rows': len(rows)}

Pass mode='Append' to add the rows to the table instead of replacing it. Large row lists are split into multiple files automatically and loaded in one operation.

Loading a file into a table

When the data is already in the lakehouse's Files section, conn.load_table turns it into a table and conn.wait_for_operation waits until the load completes.

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

# Zato
from zato.server.service import Service

class LoadOrders(Service):

    input = 'workspace_id', 'lakehouse_id'

    def handle(self):

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

        # Upload the file ..
        data = 'order_id,amount\nORD-001,250.00\nORD-002,99.90\n'.encode('utf-8')
        path = '{}/Files/incoming/orders.csv'.format(self.request.input.lakehouse_id)
        conn.onelake_write(self.request.input.workspace_id, path, data)

        # .. load it into the orders table ..
        location = conn.load_table(
            self.request.input.workspace_id,
            self.request.input.lakehouse_id,
            'orders',
            'Files/incoming/orders.csv',
        )

        # .. and wait until the load completes.
        operation = conn.wait_for_operation(location)

        self.response.payload = {'status': operation['status']}

Parquet files load with file_format='Parquet', mode='Append' adds to the table instead of replacing it, and a whole directory loads in one call with path_type='Folder' and recursive=True.

location = conn.load_table(
    workspace_id,
    lakehouse_id,
    'orders',
    'Files/incoming/orders',
    file_format='Parquet',
    mode='Append',
    path_type='Folder',
    recursive=True,
)

Listing tables

conn.list_tables returns all the tables of a lakehouse.

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

# Zato
from zato.server.service import Service

class ListTables(Service):

    input = 'workspace_id', 'lakehouse_id'

    def handle(self):

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

        tables = conn.list_tables(self.request.input.workspace_id, self.request.input.lakehouse_id)

        names = [table['name'] for table in tables]

        self.response.payload = {'tables': names}

Querying with SQL

conn.query runs an SQL query against a lakehouse and returns the rows as a list of dicts keyed by column names.

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

# Zato
from zato.server.service import Service

class GetTopRegions(Service):

    input = 'workspace_id', 'lakehouse_id'

    def handle(self):

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

        rows = conn.query(
            self.request.input.workspace_id,
            self.request.input.lakehouse_id,
            'select region, total from daily_totals where total > 1000',
        )

        for row in rows:
            self.logger.info('%s -> %s', row['region'], row['total'])

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

Queries run on a Spark session the connection opens once per lakehouse and reuses across calls, so the first query waits for Spark to start while later ones take seconds. The Livy API the session runs on needs to be enabled by your tenant admin under Tenant settings → Developer settings in the Fabric admin portal.

conn.close_spark_session closes the lakehouse's session when a service is done querying.

conn.close_spark_session(workspace_id, lakehouse_id)

Running Spark code

conn.run_spark runs arbitrary code on the same session - here, PySpark that writes a transformed table.

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

# Zato
from zato.server.service import Service

class BuildRegionSummary(Service):

    input = 'workspace_id', 'lakehouse_id'

    def handle(self):

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

        code = """
df = spark.sql('select region, sum(total) as total from daily_totals group by region')
df.write.mode('overwrite').saveAsTable('region_summary')
"""

        session_id = conn.open_spark_session(self.request.input.workspace_id, self.request.input.lakehouse_id)
        output = conn.run_spark(self.request.input.workspace_id, self.request.input.lakehouse_id, session_id, code)

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