DEV Community: Jan Giacomelli

Monitor the Performance of your Python FastAPI App with AppSignal

Jan Giacomelli — Wed, 24 Jul 2024 12:00:00 +0000

While building an app with FastAPI can be reasonably straightforward, deploying and operating it might be more challenging.

The whole user experience can be ruined by unexpected errors, slow responses, or even worse — downtime.

AppSignal is a great tool of choice for efficiently tracking your FastAPI app's performance.
It allows you to easily monitor average/95th percentile/90th percentile response times, error rates, throughput, and much more.
Useful charts are available out of the box. Let's see it in action!

What Can You Do with Performance Monitoring?

With performance monitoring, we can track app response times, throughput, error rates, CPU consumption, memory usage, etc.
Changes in these metrics can indicate something is not quite right, and we should investigate.

For example, if response times are monotonically increasing on a specific endpoint, we can investigate what's causing it. It might be inefficient code, a slow database query, a slow external API call, or something else.

In such cases, you can intervene before a gateway timeout occurs, for example, and your users start complaining.

Setting Up Our FastAPI Python Project and Configuring AppSignal

Let's use an app I've already prepared.

First, clone the repository from GitHub:

$ git clone git@github.com:jangia/fastapi_performance_with_appsignal.git
$ cd fastapi_performance_with_appsignal

Second, create a virtual environment and install the dependencies:

$ python3.12 -m venv venv
$ source venv/bin/activate
$ pip install -r requirements.txt

Note: For things to work, you need the following packages installed: opentelemetry-instrumentation-fastapi and appsignal.

Third, set the AppSignal environment variables:

$ export APPSIGNAL_PUSH_API_KEY=<your_appsignal_push_api_key>
$ export APPSIGNAL_REVISION=main

Note: You can read more about configuring AppSignal for FastAPI in Track Errors in FastAPI for Python with AppSignal.

Use environment variables to configure AppSignal in the __appsignal__.py file:

import os

from appsignal import Appsignal

appsignal = Appsignal(
    active=True,
    name="fastapi_performance_with_appsignal",
    push_api_key=os.getenv("APPSIGNAL_PUSH_API_KEY"),
    revision=os.getenv("APPSIGNAL_REVISION"),
    enable_host_metrics=True,
)

The FastAPI app looks like this:

import json
import random
import time

import requests

from appsignal import set_category, set_sql_body, set_body
from fastapi import FastAPI, Depends
from opentelemetry import trace
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from sqlalchemy.orm import Session

from __appsignal__ import appsignal
from models import SessionLocal, Task

appsignal.start()

tracer = trace.get_tracer(__name__)

app = FastAPI(
    title="FastAPI with AppSignal",
)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/hello-world")
def hello_world():
    time.sleep(random.random())
    return {"message": "Hello World"}

@app.get("/error")
def hello_world():
    raise Exception("Something went wrong. Oops!")


@app.get("/slow-external-api")
def slow_external_api():
    api_url = "http://docs.appsignal.com/"
    with tracer.start_as_current_span("Call External API"):
        set_category("external_api.http")
        set_body(json.dumps({"url": api_url}))
        response = requests.get(api_url)

    return {"message": "External API successfully called!"}


@app.get("/slow-query")
def slow_query(db: Session = Depends(get_db)):

    with tracer.start_as_current_span("List tasks"):
        query = db.query(Task)
        tasks = query.all()
        set_category("tasks.sql")
        set_sql_body(str(query))
    return {
        "tasks": [
            {"id": task.id, "title": task.title, "status": task.status}
            for task in tasks
        ],
    }


FastAPIInstrumentor().instrument_app(app)

We'll use a few endpoints to demonstrate how to monitor performance.

Monitor Response Times and Throughput in AppSignal

With our dependencies installed and environment variables set, we can start the app:

(venv)$ uvicorn main:app --reload

Once the app is up and running, we can use the call_api.py script to send requests to the app endpoints in parallel:

(venv)$ python call_api.py

This script will call every endpoint we have 20 times in parallel using asyncio.

import asyncio
from aiohttp import ClientSession


async def call_api(url: str):
    async with ClientSession() as session:
        async with session.get(url) as response:
            response = await response.text()
            print(response)


async def main():
    base_url = "http://localhost:8000"
    endpoints = ["slow-query", "slow-external-api", "hello-world", "error"]
    async with asyncio.TaskGroup() as group:
        for endpoint in endpoints:
            url = f"{base_url}/{endpoint}"
            for i in range(20):
                group.create_task(call_api(url))

asyncio.run(main())

Once the script completes, go to the AppSignal dashboard and select your application. Once on the application dashboard, choose the Performance tab -> Graphs. You'll see two charts — Response Times and Throughput:

The Response Times chart shows the average response time, 95th percentile response time, and 90th percentile response time across all endpoints inside your application.

The Throughput chart shows the number of processed requests per minute across all endpoints inside your application.

Looking at these charts is a great way to get a quick overview of how your application is performing, but it's not enough. Fortunately, AppSignal provides a way to drill into the details — you can see the same charts per endpoint.

Digging Deeper

To do that, click on Actions inside Performance:

You'll see a list of all endpoints inside your app that were called within the specified time range.

For each endpoint, you can see the average response time, 95th percentile response time, and 90th percentile response time.
You can go even deeper by clicking on the endpoint name. That will take you to the endpoint details page.

You'll see the same charts as before. This time, they are for the selected endpoint only.

You'll see another thing on the endpoint details page — errors and the error rate for the selected endpoint:

Looking at these charts, you can quickly see whether:

Response times are increasing/decreasing/stable
Throughput is increasing/decreasing/stable
Error rate is increasing/decreasing/stable

If any of these metrics show a trend in the wrong direction, you can investigate and fix them before they become a problem for your users.

Setting Alerts

While this is all great and useful, you must go to the AppSignal dashboard to see how your app is performing.
This way, you might still not react quickly enough when something goes wrong.

To overcome this, you can set an alert that triggers whenever a request takes longer than the specified threshold.

Go to the details of a /slow-query endpoint and click on View Incident.

After that, set alerts on the right side, with 5 seconds as the threshold, and select Every occurrence as Alerting:

Run the script that calls the endpoints one more time:

(venv)$ python call_api.py

You should receive an email from AppSignal warning you about the request/s that take/took too long to process. This means you don't need to constantly check the AppSignal dashboard to see how your app is performing. You'll be notified straight away if something strange happens.

Monitor Database Queries

Monitoring response times and throughput is very useful, but it only tells us a little about what's causing a problem.

As mentioned, multiple things can cause slow response times.

One is slow database queries: we'll also want to monitor them.

Let's take a look at the /slow-query endpoint:

# ... other code
tracer = trace.get_tracer(__name__)
# ... other code


@app.get("/slow-query")
def slow_query(db: Session = Depends(get_db)):

    with tracer.start_as_current_span("List tasks"):
        query = db.query(Task)
        tasks = query.all()
        set_category("tasks.sql")
        set_sql_body(str(query))
    return {
        "tasks": [
            {"id": task.id, "title": task.title, "status": task.status}
            for task in tasks
        ],
    }

Here, we're using SQLAlchemy to query the database for all tasks. To track queries inside FastAPI, we need to utilize custom instrumentation.

We can do that by executing the query inside the with tracer.start_as_current_span("List tasks"): block.
We'll create a "List tasks" span. You'll see it inside the AppSignal dashboard.

We execute the query inside the span. We need to set a category for AppSignal to recognize that this is a database query measurement using set_category("tasks.sql").

The category name must end with one of the following suffixes to be recognized as a database query:

*.active_record
*.ecto
*.elasticsearch
*.knex
*.mongodb
*.mysql
*.postgres
*.psycopg2
*.redis
*.sequel
*.sql

We also add a query string to the span body — set_sql_body(str(query)). This way, we can see the executed query inside the AppSignal dashboard.

Note: str(query) returns a query with placeholders. If you want to see a query with values, you can use str(query.statement.compile(compile_kwargs={"literal_binds": True})). But be careful not to send any sensitive data this way.

Note: A span is a single step in the execution flow.

Since you've already executed the script that calls the endpoints, you should see the /slow-query endpoint query inside the AppSignal dashboard.

Go to Slow queries under Performance:

You can click on the query name tasks.sql to see the query details. There you'll find:

The query itself (as we sent it, with set_sql_body)
Response times chart
Throughput chart

Looking at the charts, you can see the query's performance trend. If the response times are monotonically increasing, you can investigate why and fix the issue.

Hint: some possible causes for slow queries are:

Queries are not using indexes (e.g., there's a missing index or filtering is only set on non-indexed columns).
Queries are loading all the data from a database (e.g., there's missing pagination or all the data for rows is being loaded despite this being unnecessary).
N + 1 queries (e.g., querying for all users and then querying each user's tasks to satisfy a single request).
Inefficient query planning (e.g., max set on an empty result can take an abnormally long time in PostgreSQL).

Monitor Slow External API Calls

External API calls can also cause slow response times. Since external APIs are outside your control, you just have to hope for the best.

In reality, external APIs often cause problems due to being unresponsive or even down entirely.
That's why you should monitor those calls as well.

As with database queries, AppSignal has got you covered.

Let's take a look at the /slow-external-api endpoint:

@app.get("/slow-external-api")
def slow_external_api():
    api_url = "http://docs.appsignal.com/"
    with tracer.start_as_current_span("Call External API"):
        set_category("external_api.http")
        set_body(json.dumps({"url": api_url}))
        requests.get(api_url)

    return {"message": "External API successfully called!"}

Like with database queries, we need to use custom instrumentation to track external API calls.
We can do that by executing the query inside the with tracer.start_as_current_span("Call External API") block.

We set the span name to "Call External API". You'll see it inside the AppSignal dashboard.

Inside the span, we execute the external API call. Set the correct category for AppSignal to recognize that this is an external API call measurement — set_category("external_api.http").

The category name must end with one of the following suffixes to be recognized as an external API call:

*.faraday
*.grpc
*.http
*.http_rb
*.net_http
*.excon
*.request
*.requests
*.service
*.finch
*.tesla
*.fetch

We also add a called URL to the span body — set_body(json.dumps({"url": api_url})). This lets us see which URL was called inside the AppSignal dashboard.
You can add more details to the span's body by extending the dictionary you're sending to set_body. (For example, you can also add a request or response body).

Note: Ensure you don't send sensitive data inside the span's body.

Since you've already executed the script that calls the endpoints, you should see the /slow-external-api endpoint query inside the AppSignal dashboard.

Go to Slow queries under Performance:

You can click on the API call name external_api.http to see the details. There you'll find:

The URL that was called (as we sent it with set_body)
Response times chart
Throughput chart

Looking at the charts, you can see how the API call performs over time. You can decide whether to increase the timeout for the API call or change API usage (e.g., use a smaller page size or utilize a different API).

Monitor Host Metrics in AppSignal for FastAPI

Last but not least, you can monitor host metrics. AppSignal can track CPU, memory, disk, and network usage. You can enable that by setting enable_host_metrics=True inside the __appsignal__.py file:

import os

from appsignal import Appsignal

appsignal = Appsignal(
    active=True,
    name="fastapi_performance_with_appsignal",
    push_api_key=os.getenv("APPSIGNAL_PUSH_API_KEY"),
    revision="main",
    enable_host_metrics=True,  # THIS
)

So stop uvicorn, and let's run our API with docker-compose:

$ docker-compose up -d --build

Once again, run the script that calls the endpoints:

(venv)$ python call_api.py

Note: macOS/OSX is not supported. That's why we're using Docker. See AppSignal's official docs for more info.

Once the script completes, go to the AppSignal dashboard under Performance -> Host metrics — you'll see the list of hosts that are running your app:

Click on the hostname to see the details. There, you'll find these charts:

Load Average
CPU Usage
Memory Usage
Swap Usage
Disk I/O Read
Disk I/O Write
Disk Usage
Network Traffic Received
Network Traffic Transmitted

These metrics can help you determine whether your hosts have enough resources to handle the load.
You can also see whether the load is roughly equally spread across the hosts.

If you see anything that concerns you, you can investigate further and take action before it becomes a problem for your users.

And that's it!

Wrapping Up

In this post, we've seen how to monitor the performance of FastAPI applications using AppSignal.

Monitoring performance means that we can intervene before things go south and our users start complaining.

To be fully in control of our app, we should combine performance monitoring with error tracking.

Happy coding!

P.S. If you'd like to read Python posts as soon as they get off the press, subscribe to our Python Wizardry newsletter and never miss a single post!

Track Errors in FastAPI for Python with AppSignal

Jan Giacomelli — Wed, 17 Apr 2024 11:08:53 +0000

When you first try a new library or framework, you are excited about it. However, as soon as you run something on production, things are less than ideal — an error here, an exception there - bugs everywhere! You start reading your logs, but you often lack context, like how often an error happens, in what line, etc.

Fortunately, tools such as AppSignal can help. AppSignal helps you track your errors and gives you a lot of valuable insights. For example, you can very quickly see how often an error happens, in what line, and for which deployment.

That's what we'll do now: leverage AppSignal to track errors in a FastAPI application.

Setting Up Our FastAPI for Python Project

Let's prepare a project we'll use as an example. First, create a new directory and virtual environment:

$ mkdir fastapi_appsignal
$ cd fastapi_appsignal
$ python3.12 -m venv venv
$ source venv/bin/activate

Second, install FastAPI:

$ pip install "fastapi[all]"

Third, add a new file called main.py with the following contents:

import uuid
from uuid import UUID

from fastapi import FastAPI
from pydantic import BaseModel


class TodoData(BaseModel):
    title: str
    done: bool = False


class Todo(BaseModel):
    id: UUID
    title: str
    done: bool = False


TODOS = []
app = FastAPI()


@app.get("/todos")
def todo_list() -> list[Todo]:
    return TODOS


@app.post("/todos")
def todo_create(data: TodoData) -> Todo:
    todo = Todo(id=uuid.uuid4(), title=data.title, done=data.done)
    TODOS.append(todo)
    return todo


@app.post("/todos/{todo_id}")
def todo_edit(todo_id: UUID, data: TodoData) -> Todo:
    todo = next((t for t in TODOS if t.id == todo_id))
    todo.title = data.title
    todo.done = data.done
    return todo

What we have here is a very simple API example for managing TODOs. We'll use this API to demonstrate how to track errors with AppSignal.

Configure AppSignal with FastAPI

If you don't have an AppSignal account, create one - AppSignal offers a 30 day free trial.

First, we need to install appsignal and opentelemetry-instrumentation-fastapi:

$ pip install appsignal
$ pip install opentelemetry-instrumentation-fastapi

Second, let's configure AppSignal for FastAPI. Create a new file called __appsignal__.py with the following contents:

import os

from appsignal import Appsignal

appsignal = Appsignal(
    active=True,
    name="fastapi_appsignal",
    push_api_key=os.getenv("APPSIGNAL_PUSH_API_KEY"),
)

Third, update the main.py file:

import uuid
from uuid import UUID

from fastapi import FastAPI
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor  # new
from pydantic import BaseModel

from __appsignal__ import appsignal  # new

appsignal.start()  # new


class TodoData(BaseModel):
    title: str
    done: bool = False


class Todo(BaseModel):
    id: UUID
    title: str
    done: bool = False


TODOS = []
app = FastAPI()


@app.get("/todos")
def todo_list() -> list[Todo]:
    return TODOS


@app.post("/todos")
def todo_create(data: TodoData) -> Todo:
    todo = Todo(id=uuid.uuid4(), title=data.title, done=data.done)
    TODOS.append(todo)
    return todo


@app.post("/todos/{todo_id}")
def todo_edit(todo_id: UUID, data: TodoData) -> Todo:
    todo = next((t for t in TODOS if t.id == todo_id))
    todo.title = data.title
    todo.done = data.done
    return todo


FastAPIInstrumentor().instrument_app(app)  # new

We need to start the AppSignal agent alongside our FastAPI application. For more details, refer to the AppSignal Python installation docs and FastAPI instrumentation docs.

At this point, our project is ready to track our first error with AppSignal. To send errors to AppSignal, you need to get your push API key. You can get one by following the official AppSignal guide for adding a new app. After you have your push API key, start the server:

$ export APPSIGNAL_PUSH_API_KEY=<your_appsignal_push_api_key>
$ uvicorn main:app --reload

Now your server is up and running, open a new terminal and send a request to the API:

$ curl -X POST -H "Content-Type: application/json" -d '{"title": "Buy milk", "done": true}' http://localhost:8000/todos/dde18df8-7e11-4f84-bcbf-2eee1a5d157e

This request will raise the StopIteration exception inside the todo_edit function. That's because we use the next function to get the TODO with the given ID. The exception is raised since we're using the UUID of a non-existing task.

To confirm that AppSignal is working, click on Next Step at the bottom of the page from which you've copied the push API key.

Once you've clicked on Next Step, wait for AppSignal to confirm that it has received data from your app. After this, you can open the Errors -> Issue list tab and see the error we've just raised.

Congrats! You've just tracked your first error in FastAPI with AppSignal. But we can do better than that. Let's see how we can get more insights into our errors.

Get More Error Insights

At this point, we're able to track errors with AppSignal. On the dashboard, you can see:

How often an error happens
Error messages
Error backtraces

You can also add notes to errors, assign errors, and post messages to Slack when errors occur.

To get even more insights, you can set up deploy markers and backtrace links.

Using them together lets you jump directly to the source code on your GitHub repository from the error's backtrace. For this to work, you need to push your code to GitHub. Do that before going further (you can follow the official guide). Once this is done, we can continue setting up deploy markers and backtrace links.

Deploy Markers

First, we need to set up deploy markers. To do that, update the __appsignal__.py file:

import os

from appsignal import Appsignal

appsignal = Appsignal(
    active=True,
    name="fastapi_appsignal",
    push_api_key=os.getenv("APPSIGNAL_PUSH_API_KEY"),
    revision=os.getenv("APPSIGNAL_REVISION"),  # new
)

We enable deploy markers by setting the revision value to a non-null value. Setting up deploy markers will allow us to see which deployment causes an error. We'll set APPSIGNAL_REVISION to main because that's the name of our branch. In the production setup, you should set this to the commit hash of the deployed version or tag name.

Second, we need to enable backtrace links. We already enabled deploy markers, so we're halfway there. What's left is to link our GitHub repository. To do that, go to your organization's settings and click on Install GitHub App.

Once on GitHub, select your organization.

After that, keep everything as it is and click Install.

You'll be redirected back to AppSignal's dashboard. Select the repository you want to link to your app and click Save repository selection.

Let's raise another error to test that everything is working as expected. To do that, set environment variables and start your server again:

$ export APPSIGNAL_PUSH_API_KEY=<your_appsignal_push_api_key>
$ export APPSIGNAL_REVISION=main
$ uvicorn main:app --reload

After that, open a new terminal and send another request to the API:

$ curl -X POST -H "Content-Type: application/json" -d '{"title": "Buy milk", "done": true}' http://localhost:8000/todos/dde18df8-7e11-4f84-bcbf-2eee1a5d157e

Backtrace Links

Once the latest error is tracked, you can try backtrace links. Go to the Errors -> Issue list tab and click on the error we've just raised. Once on the error's detail page, click _Inspect latest example.

You'll end up on the details of the latest error sample, where you'll see the Backtrace section. On the right side of the backtrace, you'll see a link to the source code on GitHub - git.

When you click on git, you'll be redirected to the source code on GitHub. You'll see the exact line where the error happened. Now, you can easily pinpoint the problematic code and fix the issue. This is way less work than parsing log files and doing things manually.

Note: If you're not pointed to the correct line of code on GitHub, make sure you've committed and pushed the latest version of your code to GitHub.

Reporting Python FastAPI Errors Explicitly

So far, we've seen how to track errors raised by our FastAPI application. But we might want to report errors explicitly in some cases. For example, we might use an AI API to generate emails for our clients.

As you may know, these APIs are often unreliable, and seeing internal server errors is quite common. In such cases, we want to handle the error - for example, by falling back to a default template.

We want to report an error to AppSignal to see how often it happens and what the error message is.

Using `set_error`

In some cases, we can make usage more robust (e.g., by implementing retries) or contact the API provider's support.
To handle such scenarios, AppSignal provides the set_error function. Let's see how to use it.

Update your main.py file:

# ... existing imports
from appsignal import set_error  # new
from fastapi import FastAPI, HTTPException  # new
# ... existing imports


# ... existing code

@app.post("/todos/{todo_id}")
def todo_edit(todo_id: UUID, data: TodoData) -> Todo:
    try:  # new
        todo = next((t for t in TODOS if t.id == todo_id))
    except StopIteration:
        set_error(Exception("Todo not found"))
        raise HTTPException(status_code=404, detail="Todo not found")
    todo.title = data.title
    todo.done = data.done
    return todo


# ... existing code

In our case, users won't see the Internal server error anymore. Instead, we'll return the Todo not found error.
That's better for our users and our API. Without set_error, nothing would be reported to AppSignal. With set_error, the error is reported and tracked as before. Hence, we'll see how often this error happens and the error message. Backtrace links still apply, as in the previous example.

To test that everything is working as expected, follow the same steps as above:

Start the server.
Send a request to the API.
Go to the dashboard Errors -> Issue list in AppSignal and check the new error.

That's all great, but sometimes, we need more context when manually reporting an error to AppSignal.
In such cases, we can use the send_error_with_context function. This function allows us to manually add needed context to the error - e.g., the ID of a non-existing task.
Once again, update the main.py file:

# ... existing imports
from appsignal import send_error_with_context, set_params  # new
from fastapi import FastAPI, HTTPException  # new
# ... existing imports


# ... existing code

@app.post("/todos/{todo_id}")
def todo_edit(todo_id: UUID, data: TodoData) -> Todo:
    try:
        todo = next((t for t in TODOS if t.id == todo_id))
    except StopIteration:
        with send_error_with_context(Exception("Todo not found")):
            set_params({"todo_id": todo_id})
        raise HTTPException(status_code=404, detail="Todo not found")
    todo.title = data.title
    todo.done = data.done
    return todo



# ... existing code

To test that everything works as expected, follow the same steps as above.

In this case, you'll also see the Parameters section that contains the parameters you've explicitly set (here, that's the ID of the non-existing task).

And that's it!

Wrapping Up

In this blog post, we've seen how to track errors in FastAPI with AppSignal. We set up AppSignal for FastAPI, tracked errors, and gained some further insights.

Besides that, we also learned how to explicitly track errors using the AppSignal functions set_error and send_error_with_context. This should give you a good starting point to track your errors better and fix them faster from now on.

Happy engineering!

P.S. If you'd like to read Python posts as soon as they get off the press, subscribe to our Python Wizardry newsletter and never miss a single post!

DEV Community: Jan Giacomelli

Monitor the Performance of your Python FastAPI App with AppSignal

What Can You Do with Performance Monitoring?

Setting Up Our FastAPI Python Project and Configuring AppSignal

Monitor Response Times and Throughput in AppSignal

Digging Deeper

Setting Alerts

Monitor Database Queries

Monitor Slow External API Calls

Monitor Host Metrics in AppSignal for FastAPI

Wrapping Up

Track Errors in FastAPI for Python with AppSignal

Setting Up Our FastAPI for Python Project

Configure AppSignal with FastAPI

Get More Error Insights

Deploy Markers

Backtrace Links

Reporting Python FastAPI Errors Explicitly

Using set_error

Wrapping Up

Using `set_error`