Introducing Python functions on Hasura DDN

Hasura Data Delivery Network (DDN) simplifies backend development by offering unparalleled composability across data sources through its supergraph architecture. Hasura DDN enables engineering teams to effortlessly deliver a unified API that exposes all your data sources through a single GraphQL endpoint.

The supergraph architecture of Hasura DDN allows you to define relationships between different data sources, and now with your code as well. Code introspection can reduce the initial overhead of manual schema generation.

With the support of Python functions on Hasura DDN, you can focus on writing sync or async functions that return some data. Hasura will do the code introspection and generate all the configurations needed for a production-ready GraphQL API – including all configurations for the GraphQL schema, observability readiness, and more advanced configurations.

Hasura utilizes the Python runtime and supports installing packages from PyPi via pip allowing you to use your favorite Python libraries. You can easily make use of pandas and NumPy for data analysis, or use Matplotlib to create beautiful charts, then utilize requests to upload the charts to an S3 bucket.

In this blog post, we’ll explore how you can use Python functions to create custom queries and mutations on your supergraph. With this approach, you can:

• Integrate custom business logic into Hasura DDN.
• Perform data transformations and integrate with external APIs.
• Access data from unsupported data sources by writing custom code.

Hasura DDN introduces Python functions capable of handling GraphQL queries and mutations. This is fantastic for developers because Hasura Python functions allow you to:

• Not worry about resolvers and abstract away many of the complexities of using GraphQL, just simply write functions.
• Use synchronous or asynchronous Python functions that get turned into GraphQL queries and mutations.
• Monitor and trace function executions for performance optimization and debugging with built-in OpenTelemetry support.
• Customize and extend telemetry data with extensible spans to gain deeper insights into GraphQL API performance.
• Use any Python libraries within your functions, letting you bring the tools you know and love with you into the Hasura ecosystem.

The Python Lambda Connector implements the Hasura data connector specification using the Python NDC SDK and provides the user with a `FunctionConnector` that supplies decorators for registering queries and mutations. When you first initialize the connector, you will see some default functions provided in the functions.py file as examples showing you how to use the connector.

In the simplest example, there’s a function called hello that accepts a name as an argument and returns a string output.

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnector

connector = FunctionConnector()

# This is an example of a simple function that can be added onto the graph
@connector.register_query # This is how you register a query
def hello(name: str) -> str:
    return f"Hello {name}"

if __name__ == "__main__":
    start(connector)

When Hasura introspects the function, it generates metadata to expose the function as a command on the graph. This declarative metadata makes it easy to change things like the name of a field. This also makes it easy to enforce organizational guidelines for things like permissions with the use of custom linter rules.

---
kind: Command
version: v1
definition:
  name: Hello
  outputType: String!
  arguments:
    - name: name
      type: String!
  source:
    dataConnectorName: python
    dataConnectorCommand:
      function: hello
  graphql:
    rootFieldName: python_hello
    rootFieldKind: Query

You will be able to execute this query with the following GraphQL:

query MyQuery {
  python_hello(name: "Tristen")
}

If you wanted to join data with this function, you could easily add a relationship mapping a field in a collection to the function argument. Imagine you had a user collection inside a PostgreSQL database and you wanted to expose the python_hello command so you could generate a hello message with the user's name. You could add a relationship as follows:

---
kind: Relationship
version: v1
definition:
 name: helloMessage
 sourceType: User
 target:
   command:
     name: Hello
     subgraph: python
 mapping:
   - source:
       fieldPath:
         - fieldName: name
     target:
       argument:
         argumentName: name

Then you could execute this query to perform a join from the database-backed user collection to your Python function.

query MyQuery{
  user {
    name
    helloMessage
  }
}

As you can imagine, this is extremely powerful and allows you to enrich your data in ways previously not possible.

If you have tried Hasura v2 in the past, joining data from databases, remote schemas, or even actions to other actions was not supported. Hasura DDN now inherently supports this capability.

Hasura DDN offers built-in support for seamless integration with OpenTelemetry, enabling you to monitor and trace Lambda function executions without additional configurations. You can add custom tracing spans and span attributes for detailed tracing. This enhanced observability will help you optimize performance, identify bottlenecks, and troubleshoot issues more effectively.

In the template code, you will find an example of how to create custom spans.

from hasura_ndc import start
from hasura_ndc.instrumentation import with_active_span
from opentelemetry.trace import get_tracer
from hasura_ndc.function_connector import FunctionConnector

connector = FunctionConnector()
tracer = get_tracer("ndc-sdk-python.server")

# Utilizing with_active_span allows the programmer to add Otel tracing spans
@connector.register_query
async def with_tracing(name: str) -> str:

    def do_some_more_work(_span, work_response):
        return f"Hello {name}, {work_response}"

    async def the_async_work_to_do():
        # This isn't actually async work, but it could be! Perhaps a network call belongs here, the power is in your hands fellow programmer!
        return "That was a lot of work we did!"

    async def do_some_async_work(_span):
        work_response = await the_async_work_to_do()
        return await with_active_span(
            tracer,
            "Sync Work Span",
            lambda span: do_some_more_work(span, work_response), # Spans can wrap synchronous functions, and they can be nested for fine-grained tracing
            {"attr": "sync work attribute"}
        )

    return await with_active_span(
        tracer,
        "Root Span that does some async work",
        do_some_async_work, # Spans can wrap asynchronous functions
        {"tracing-attr": "Additional attributes can be added to Otel spans by making use of with_active_span like this"}
    )

if __name__ == "__main__":
    start(connector)

You’ll be able to see these traces directly from the Hasura Console:

Make use of the Pydantic library to create typed parameters and responses. Unleash the power of GraphQL.

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnector
from pydantic import BaseModel

connector = FunctionConnector()

class Pet(BaseModel):
    name: str
    
class Person(BaseModel):
    name: str
    pets: list[Pet] | None = None

class GreetingResponse(BaseModel):
    person: Person
    greeting: str

@connector.register_query
def greet_person(person: Person) -> GreetingResponse:
    greeting = f"Hello {person.name}!"
    if person.pets is not None:
        for pet in person.pets:
            greeting += f" And hello to {pet.name}.."
    else:
        greeting += f" I see you don't have any pets."
    return GreetingResponse(
        person=person,
        greeting=greeting
    )

if __name__ == "__main__":
    start(connector)

Now you can issue this GraphQL query:

query Q {
  python_greetPerson(
    person: {name: "Tristen", pets: [{name: "Whiskers"}, {name: "Smokey"}]}
  ) {
    greeting
    person {
      name
    }
  }
}

And get a response:

{
  "data": {
    "python_greetPerson": {
      "greeting": "Hello Tristen! And hello to Whiskers.. And hello to Smokey..",
      "person": {
        "name": "Tristen"
      }
    }
  }
}

Error handling in server-side functions involves managing error messaging across various channels, such as API responses, admin alerts, and logging for debugging purposes. By default, unhandled errors are caught by the Lambda SDK host, resulting in an InternalServerError response to Hasura. While internal error details are logged in the OpenTelemetry trace, GraphQL API clients receive a generic "internal error" response.

To return specific error details to GraphQL API clients, developers can deliberately raise predefined error classes provided by the Lambda SDK, such as Forbidden, Conflict, and UnprocessableContent.

For example, you can return a UnprocessableContent error to the client as follows:

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnectorfrom hasura_ndc.errors import UnprocessableContent

connector = FunctionConnector()

@connector.register_query
def error():
    raise UnprocessableContent(message="This is an error", details={"Error": "This is an error!"})

if __name__ == "__main__":
    start(connector)

The API response would be:

{
  "data": null,
  "errors": [
    {
      "message": "error from data source: This is an error",
      "path": [
        "python_error"
      ],
      "extensions": {
        "details": {
          "Error": "This is an error!"
        }
      }
    }
  ]
}

For I/O bound tasks when joining from a collection to a function, it’s possible to specify a degree of parallelism to control the maximum number of concurrent executions for a function for each query.

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnector
import asyncio

connector = FunctionConnector()

# This is an example of how to set up queries to be run in parallel for each query
@connector.register_query(parallel_degree=5) # When joining to this function, it will be executed in parallel in batches of 5
async def parallel_query(name: str) -> str:
    await asyncio.sleep(1)
    return f"Hello {name}"

if __name__ == "__main__":
    start(connector)

Only queries can be executed in parallel, making the parallel_degree parameter invalid when registering mutations.

Create charts with Matplotlib

You can bring your favorite Python libraries with you to Hasura DDN, just add them to the requirements.txt file and use them as you normally would. For example, you can create a function that will generate a bar chart and return it as a base64 encoded string.

First, you’d add Matplotlib to the requirements.txt file and install the library with pip.

matplotlib==3.9.1.post1

Then you could write a function that generates a bar chart.

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnector
import matplotlib
# Use the non-interactive backend
matplotlib.use('Agg')
import matplotlib.pyplot as plt
import io
import base64

connector = FunctionConnector()

@connector.register_query
def base64_bar_chart(labels: list[str], values: list[float], title: str, xlabel: str, ylabel: str) -> str:
    plt.figure(figsize=(10, 6))
    plt.bar(labels, values)
    plt.title(title)
    plt.xlabel(xlabel)
    plt.ylabel(ylabel)
    if len(labels) > 5:
        plt.xticks(rotation=45, ha='right')
    buffer = io.BytesIO()
    plt.savefig(buffer, format='png')
    buffer.seek(0)
    image_base64 = base64.b64encode(buffer.getvalue()).decode('utf-8')
    plt.close()
    return image_base64

if __name__ == "__main__":
    start(connector)

Once you have tracked this function you can run the query:

query MyQuery {
  python_base64BarChart(
    labels: ["A", "B", "C"]
    title: "Hello"
    values: [1.5, 3.5, 1.2]
    xlabel: "X"
    ylabel: "Y"
  )
}

And you’ll get back the chart encoded as a base64 string.

Send requests to other APIs or services

You can send requests with the popular requests library to pull external APIs or services into your graph allowing you to join together your data with the APIs you use. This is a powerful way to enrich the capabilities of existing APIs. For example, you can easily add pagination to an API that doesn’t support it:

from hasura_ndc import start
from hasura_ndc.function_connector import FunctionConnector
from pydantic import BaseModel
import requests

class Posts(BaseModel):
    userId: int
    id: int
    title: str
    body: str

@connector.register_query
def fetch_posts(limit: int | None = None, offset: int | None = None) -> list[Posts]:
    response = requests.get('https://jsonplaceholder.typicode.com/posts')
    if response.status_code == 200:
        posts_data = response.json()

        if offset:
            posts_data = posts_data[offset:]
        
        if limit:
            posts_data = posts_data[:limit]
            
        return [Posts(**post) for post in posts_data]
    else:
        raise Exception(f"Failed to fetch posts. Status code: {response.status_code}")

if __name__ == "__main__":
    start(connector)

Now you can issue a query and the response will be processed inside the connector reducing the load placed on your client application with the added pagination.

query MyQuery {
  python_fetchPosts(offset: 5, limit: 5) {
    id
    title
    userId
  }
}

This powerful addition to the Hasura toolbox gives developers even more flexibility to implement complex business rules, perform data transformations, and integrate with external services directly within their GraphQL API. This streamlined approach eliminates the overhead of managing additional infrastructure, resulting in faster development cycles and reduced operational complexity.

Simplify your development process today, and take advantage of a familiar programming model that can seamlessly integrate with your existing data sources. You can focus on building core business logic while leveraging the scalability, observability, and performance of the Hasura Data Delivery Network.

If you want to learn more about the Python Lambda Connector join our next Community Call on August 29th.

Ready to take Hasura DDN for a spin? Simply click here to start your journey toward a more streamlined, modern approach to data architecture and access!