Getting data from an API using Python notebook

Getting data from an API using Python notebook

A no-code way to get data from an API is by using a Data Factory pipeline, syncing the response to a Lakehouse, or maybe transforming the response and syncing its output to a Lakehouse.

If the API call is simple, you may wish to use a python notebook to make the call and write the table.

For this you can use the requests module

import requests

import pandas as pd

key = "9b5c209c579fdba640a0ea31da3cb4ae"

lat = 53.3952792

lon = -2.1091572

url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={key}"

try:

    response = requests.get(url)  

    response.raise_for_status()

    # Show content (which is a byte array)

    print (response.content)

    # Get a section from the contect and represent it as a table

    df_items = pd.json_normalize(response.json()['main'])

    display (df_items)    

    # If we ommit the section filter: e.g. df_items = pd.json_normalize(response.json())

    # we would get columns with dot.notation in the name

    # e.g. coord.lon, coord.lat, main.temp, clouds.all

# Exception handling order is important as some are hierarchical

except requests.exceptions.ConnectionError as err:

    print ("ConnectionError:",err)

except requests.exceptions.Timeout as err:

    print ("Timeout:",err)

except requests.exceptions.TooManyRedirects as err:

    print ("TooManyRedirects:",err)

except requests.exceptions.HTTPError as err:

    print ("HTTPError:",err)

except requests.exceptions.RequestException as err:

    print ("RequestException: Something Else",err)

We can call the fabric api from a fabric notebook without authentication by using the sempy.FabricRestClient module

This also gives us access ot the default lakehouse

FabricRestClient doesn't have aexception handling but we can add it from the sempy.exceptions module

import sempy.fabric as fabric

from sempy.fabric.exceptions import FabricHTTPException, WorkspaceNotFoundException

import pandas as pd

client = fabric.FabricRestClient()

def get_workspace_items(workspace=None):

    df = None

    if workspace is None:

        workspaceId = fabric.get_workspace_id()

    else:

        workspaceId = fabric.resolve_workspace_id(workspace)

    try:

        response = client.get(f"/v1/workspaces/{workspaceId}/items")

        if response.status_code != 200:

            raise FabricHTTPException(response)

        df = pd.json_normalize(response.json()['value'])

    except WorkspaceNotFoundException as e:

        print("Caught a WorkspaceNotFoundException:", e)

    except FabricHTTPException as e:

        print("Caught a FabricHTTPException. Check the API endpoint, authentication.")

    return df

# Get the workspace items from the workspace named "Dev"

df_items = get_workspace_items("Dev")

# Get the workspace items from the current workspace, regardless of its name

df_items = get_workspace_items()

POST request

For both request and fabricRestClient, issuing a post is simply a case of calling .post,

rather than .get, and adding a parameter contaiing the content to be posted

A Fabric post request...

Here I put the status handler outside of the function as the post doesn't necessarily return a 201, and I wanted to make the post function generic.

import sempy.fabric as fabric

from sempy.fabric.exceptions import FabricHTTPException, WorkspaceNotFoundException

client = fabric.FabricRestClient()

def post_operation(payload:dict, workspace=None):

    if workspace is None:

        workspaceId = fabric.get_workspace_id()

    else:

        workspaceId = fabric.resolve_workspace_id(workspace)

    try:

        response = client.post(f"/v1/workspaces/{workspaceId}/items",json= payload)

        return response.status_code

    except WorkspaceNotFoundException as e:

        print("Caught a WorkspaceNotFoundException:", e)

    except FabricHTTPException as e:

        print("Caught a FabricHTTPException. Check the API endpoint, authentication.")

# Test API post operation by creating a lakehouse object

# Create the payload/body for the POST request as a dict

# example; create a lakehouse in the default (current) workspace

payload = {"displayName": "generated_lakehouse_test","type": "Lakehouse"}

status_code = post_operation(payload=payload, workspace=None)

if status_code != 201:

    raise FabricHTTPException(response)

Issue a Kusto query to an eventhouse.

We will need a token for this, using the notebookutils.credentials module

token = notebookutils.credentials.getToken(eventhouse_connection)

This isn't part of the request, so we obtain it outside of the query function and pass it as a parameter

import requests

import json

import pandas as pd

from datetime import datetime

def query_kusto(cluster_url: str, database: str, query: str, token: str):

    """

    Function to query Fabric Kusto db using REST API

    Args:

        cluster_url: Full cluster URL (e.g., 'https://guid.kusto.fabric.microsoft.com')

        database: Database name

        query: KQL query

        token: Bearer token

    """

    headers = {

        'Authorization': f'Bearer {token}',

        'Content-Type': 'application/json',

        'Accept': 'application/json'

    }

    body = {

        'db': database,

        'csl': query

    }

    response = requests.post(

        f'{cluster_url}/v1/rest/query',

        headers=headers,

        json=body

    )

    return response.json()

# Example call

I created an eventhouse with a table created from Microsoft's Bikes sample file.

eventhouse_connection = f'https://xxxxxxxxxx.zz.kusto.fabric.microsoft.com'

database = 'eh_DBTL'

kqlquery = '''

Bikes | take 2

'''

# Get a token to pass to the query function

token = notebookutils.credentials.getToken(eventhouse_connection)

try:

    results = query_kusto(eventhouse_connection, database, kqlquery, token)

    if results.get('Tables'):

        rows = results['Tables'][0]['Rows']

        for row in rows:

            print(row)

# We should add some specific exception handlers

except Exception as e:

    print(f"Error querying Kusto: {str(e)}")

Exception handling

We should hadle specific exceptions differently and have a catch all, plus we can add a custom exception handler.

Here we call the requests module with a ficticious url that will fail

This would normally trigger an HTTPError

import requests

class CustomException(Exception):

    pass

try:

    response = requests.get("https://fjdjdfdfd.com")

    if response.status_code != 200:

        raise CustomException("Unexpected status code: " + str(response.status_code))

   

    print("Request successful!")

   

except CustomException as e:

    print("Custom Exception:", e)

except requests.exceptions.RequestException as e:

    print("An error occurred:", e)