Index

June 6, 2024
in FastAPI, Starlette, Uvicorn
5 min read

Understanding client disconnection in FastAPI

This blog post will give you a comprehensive understanding how FastAPI works when the client disconnects.

Info

If you want to try the code I'll be presenting, you'll need to have some packages installed:

pip install httpx fastapi uvicorn httptools uvloop

httpx is going to be used as the HTTP client.
uvicorn is the ASGI server, and httptools and uvloop are packages used by uvicorn.
You know about fastapi... But it's an ASGI web framework.

A Simple Request

Let's create a FastAPI application with an endpoint that takes a long time to finish its processing.

The following endpoint just sleeps for 10 seconds, and sends a response with a 204 status code:

main.py

from fastapi import FastAPI, Request

app = FastAPI()


@app.get("/", status_code=204)
async def home() -> None:
    await anyio.sleep(10)

You can run this application with any ASGI server. We'll use Uvicorn because I maintain it, and it's the most popular ASGI server.

Let's run it with uvicorn main:app --reload --log-level=trace.

Tip

The --log-level=trace is used to see ASGI messages, and changes in the connection status.

Let's call this endpoint with an HTTP client, and disconnect before the server is able to send the response back.

client.py

import anyio
import httpx

async def main() -> None:
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        with anyio.fail_after(1):
            await client.get("/")

if __name__ == "__main__":
    anyio.run(main)

If you run the above with python client.py, you'll see the following logs on the server side:

TRACE:    127.0.0.1:50953 - HTTP connection made
TRACE:    127.0.0.1:50953 - ASGI [2] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'}, 'http_version': '1.1', 'server': ('127.0.0.1', 8000), 'client': ('127.0.0.1', 50953), 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'state': {}, 'method': 'GET', 'path': '/', 'raw_path': b'/', 'query_string': b''}
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.request', 'body': '<0 bytes>', 'more_body': False}
TRACE:    127.0.0.1:50953 - HTTP connection lost
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.disconnect'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.start', 'status': 204, 'headers': '<...>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.body', 'body': '<0 bytes>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Completed

This may look a bit too complex, so let's go step by step here...

The first two lines show that the client connected to the server, and that the ASGI application was called.

TRACE:    127.0.0.1:50953 - HTTP connection made
TRACE:    127.0.0.1:50953 - ASGI [2] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'}, 'http_version': '1.1', 'server': ('127.0.0.1', 8000), 'client': ('127.0.0.1', 50953), 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'state': {}, 'method': 'GET', 'path': '/', 'raw_path': b'/', 'query_string': b''}
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.request', 'body': '<0 bytes>', 'more_body': False}
TRACE:    127.0.0.1:50953 - HTTP connection lost
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.disconnect'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.start', 'status': 204, 'headers': '<...>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.body', 'body': '<0 bytes>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Completed

Info

The ASGI specification determines how the server and the web framework are going to interact to process the client's request.

When the server reads the body, it will send a http.request ASGI message to the ASGI application (in this case, FastAPI):

TRACE:    127.0.0.1:50953 - HTTP connection made
TRACE:    127.0.0.1:50953 - ASGI [2] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'}, 'http_version': '1.1', 'server': ('127.0.0.1', 8000), 'client': ('127.0.0.1', 50953), 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'state': {}, 'method': 'GET', 'path': '/', 'raw_path': b'/', 'query_string': b''}
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.request', 'body': '<0 bytes>', 'more_body': False}
TRACE:    127.0.0.1:50953 - HTTP connection lost
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.disconnect'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.start', 'status': 204, 'headers': '<...>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.body', 'body': '<0 bytes>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Completed

Then... Before the application finishes the execution, the client disconnected!

TRACE:    127.0.0.1:50953 - HTTP connection made
TRACE:    127.0.0.1:50953 - ASGI [2] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'}, 'http_version': '1.1', 'server': ('127.0.0.1', 8000), 'client': ('127.0.0.1', 50953), 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'state': {}, 'method': 'GET', 'path': '/', 'raw_path': b'/', 'query_string': b''}
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.request', 'body': '<0 bytes>', 'more_body': False}
TRACE:    127.0.0.1:50953 - HTTP connection lost
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.disconnect'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.start', 'status': 204, 'headers': '<...>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.body', 'body': '<0 bytes>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Completed

The server notices, and sends a http.disconnect message to the application.

TRACE:    127.0.0.1:50953 - HTTP connection made
TRACE:    127.0.0.1:50953 - ASGI [2] Started scope={'type': 'http', 'asgi': {'version': '3.0', 'spec_version': '2.4'}, 'http_version': '1.1', 'server': ('127.0.0.1', 8000), 'client': ('127.0.0.1', 50953), 'scheme': 'http', 'root_path': '', 'headers': '<...>', 'state': {}, 'method': 'GET', 'path': '/', 'raw_path': b'/', 'query_string': b''}
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.request', 'body': '<0 bytes>', 'more_body': False}
TRACE:    127.0.0.1:50953 - HTTP connection lost
TRACE:    127.0.0.1:50953 - ASGI [2] Receive {'type': 'http.disconnect'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.start', 'status': 204, 'headers': '<...>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Send {'type': 'http.response.body', 'body': '<0 bytes>'}
TRACE:    127.0.0.1:50953 - ASGI [2] Completed

Ok... Cool! Now we finally arrive to the important point of this blog post!

The client disconnected, the ASGI server communicated it to the application, but... Did the application stop?

The answer is: NO. Although the application is able to check the http.disconnect, Starlette only does it for StreamingResponses, but it doesn't do it for all the other response classes by default.

Check Client Disconnection

I said above that the application is able to check, but it doesn't do it by default.

I'll teach you how you can check when a client is disconnected.

Note

The current way to check client disconnection is a bit complicated. But... We are working on new mechanism that will be introduce in a future release with the goal of simplifying this flow.

Follow me on LinkedIn and Twitter, and sponsor me on GitHub for more information.

Let's complicate a bit our application... I'll explain everything, don't worry.

main.py

from fastapi import FastAPI, Request
import anyio
import httpx

app = FastAPI()


async def disconnected(request: Request, cancel_scope: anyio.CancelScope) -> None:
    while True:
        message = await request.receive()
        if message["type"] == "http.disconnect":
            cancel_scope.cancel()
            break


@app.get("/", status_code=204)
async def home(request: Request) -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(disconnected, request, tg.cancel_scope)
        await anyio.sleep(10)

Cool! But... What have I done?

We've created the disconnected() task, that will await on request.receive(), and cancel the anyio.TaskGroup when the message "http.disconnect" is found.

Is the logic above 100% correct? When I was writing this article, I actually thought it was, but then I remembered that I forgot a small detail... What if the client doesn't disconnect?

Well... Then the task runs forever. So yeah, we need to actually stop the TaskClient when either: 1. the client disconnects or... 2. the endpoint finishes to process its logic, and is ready to send the response.

The right logic is a bit more complex, but would be...

main.py

from typing import Any, Awaitable

import anyio
import httpx
from fastapi import FastAPI, Request

app = FastAPI()

async def disconnected(request: Request) -> None:
    while True:
        message = await request.receive()
        if message["type"] == "http.disconnect":
            break  # (1)!


async def wrap(call: Awaitable[Any], cancel_scope: anyio.CancelScope):
    await call
    cancel_scope.cancel()  # (2)!


@app.get("/", status_code=204)
async def home(request: Request) -> None:
    async with anyio.create_task_group() as tg:
        tg.start_soon(wrap, disconnected(request), tg.cancel_scope)
        await wrap(anyio.sleep(5), tg.cancel_scope)

We removed the cancel_scope.cancel() from here.
We added the cancel_scope.cancel() in the wrap() function.

Now, we achieving our goal. You can try calling the python client.py, and you'll see it will work. You can also call the endpoint with a simple curl http://localhost:8000/ (without disconnecting).

After seeing all of the above, you may have some questions...

Is this necessary?

I don't recommend to do it in most of cases. I'm just presenting a behavior, and explaining how to overcome it with the current mechanisms that are available.

Is this the best way to do this?

For now, yes. As I said above, we are working on a new mechanism to detect if the client has disconnected.

What about WebSockets?

If there's curiosity, I'll write a blog post about it as well. There are some subtle (but important) differences.

Conclusion

If you learned something useful with this blog post, consider [sponsoring me on GitHub], and/or share this blog post among your colleagues.

If you have more ideas about what would be interesting to share, feel free to let me know on LinkedIn or Twitter.

May 28, 2024
in Uvicorn
5 min read

Uvicorn 0.30.0 Release

Today, we are releasing Uvicorn 0.30.0!

There were several changes here, including:

Deprecate the uvicorn.workers module #2302
Add a new multiprocess manager #2183
Allow ConfigParser or a io.IO[Any] on log_config #1976

Deprecate the `uvicorn.workers`

The uvicorn.workers module is used to provide two classes: UvicornWorker and UvicornH11Worker.

These classes are used to run Uvicorn with Gunicorn e.g. gunicorn -k uvicorn.workers.UvicornWorker -w 4 main:app.

Gunicorn is a popular WSGI server that can run multiple worker processes to handle incoming requests. When used with Uvicorn, Gunicorn would act as a process manager and Uvicorn would act as the server handling the requests.

However, this approach is not recommended anymore. In this release, we also introduced a new multiprocess manager on Uvicorn's side, that is meant to replace Gunicorn entirely.

For backward compatibility, you can install the Uvicorn Worker package:

pip install uvicorn-worker

Add a new multiprocess manager

A new multiprocess manager was added to Uvicorn. The main goal is to be able to have a proper process manager that can handle multiple workers and restart them when needed.

This was a long-awaited feature, and it's finally here!

Nothing needs to be done from the users side, the changes are already in place when using the --workers parameter.

You can see more details on the the pull request #2183.

Allow `ConfigParser` or a `io.IO[Any]` on `log_config`

With this change, you can now pass a ConfigParser programmatically to the log_config parameter on the Uvicorn class.

import configparser
from uvicorn import Config, Server

config = Config(app=..., log_config=configparser.ConfigParser())
server = Server(config)

May 12, 2024
in WebSockets
5 min read

Managing Exceptions in WebSockets with FastAPI

In this post, we delve into the management of exceptions in WebSockets, focusing on a potent but often overlooked feature: the WebSocketException offered by Starlette.

Understanding `WebSocketException`

Conceptually, WebSocketException enables you to close a WebSocket connection with a specific code and reason by raising this exception in your WebSocket route. Here's an illustrative example:

from fastapi import FastAPI, WebSocket
from fastapi.exceptions import WebSocketException

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    raise WebSocketException(code=1008, reason="Closing the connection...")

In this instance, a WebSocketException is raised bearing the code 1008 and the explicit reason for closure: "Closing the connection...".

To run this application, first install FastAPI, Uvicorn, and WebSockets:

pip install fastapi uvicorn websockets

Run the application using Uvicorn:

uvicorn main:app --reload

On opening a WebSocket connection to ws://localhost:8000/ws, you will find the connection being closed with focal code 1008 and the attributed reason.

I use wscat for testing WebSocket connections; you can install it with the following command:

npm install -g wscat

A connection is opened with:

wscat -c ws://127.0.0.1:8000/ws

Handle custom exceptions

The app.exception_handler can be used to handle custom exceptions. Consider the following sample:

from fastapi import FastAPI, WebSocket
from fastapi.exceptions import WebSocketException

app = FastAPI()

class CustomException(Exception): ...

@app.exception_handler(CustomException)
async def custom_exception_handler(websocket: WebSocket, exc: Exception):
    await websocket.close(code=1008, reason="Custom exception")

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    raise CustomException()

In this example, the client receives a WebSocket closure with code 1008 and the reason "Custom exception" when the CustomException is raised.

Feel free to connect with me on LinkedIn for any questions or discussions on this topic.

Happy coding! 🚀

March 14, 2023
in Testing
3 min read

Contract Testing with HTTPX - Part 2

Note

This is a continuation of Contract Testing with HTTPX.

On the previous article, I used RESPX to call the service B from service A. Although it looks cool, we can actually achieve the same goal without using anything besides FastAPI itself.

The Services

Let's assume that we have similar two services as presented in the previous article.

The difference here is that we'll be creating a dependency called service_b_client, which is going to return a httpx.AsyncClient instance that calls our service B.

Service AService BRun

service_a.py
from typing import AsyncIterator

import httpx
from fastapi import APIRouter, Depends, FastAPI

router = APIRouter(prefix="/a")


async def service_b_client() -> AsyncIterator[httpx.AsyncClient]:
    async with httpx.AsyncClient(base_url="http://localhost:8000/b/") as client:
        yield client


@router.get("/")
def get_a():
    return {"a": "a"}


@router.get("/call_b")
async def call_b(client: httpx.AsyncClient = Depends(service_b_client)):
    response = await client.get("/")
    return response.json()


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8001)  # (1)!

The port is 8001, not 8000, to avoid conflicts with the other service.

service_b.py
from fastapi import APIRouter, FastAPI

router = APIRouter(prefix="/b")


@router.get("/")
def get_b():
    return {"b": "b"}


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8000)

Install the dependencies:

python -m pip install uvicorn fastapi httpx

Then open the terminal and run:

python service_a.py

Then open another terminal, and run:

python service_b.py

Now, let's call the /a/call_b endpoint.

http :8001/a/call_b # (1)!

The HTTP client used is called HTTPie, but you can use [curl], or just go to the browser, and access http://localhost:8001/a/call_b.

The response should look like:

{
    "b": "b"
}

Testing

Since the only difference between this article, and the previous one is the creation of the dependency on service A, you might be guessing that we are going to override the dependency, and... You are right! (if you didn't, is fine as well )

We are going to use app.dependency_overrides to override the service_b_client dependency, and instead of calling the real service B, we'll call the application itself, avoiding the network calls that would potentially slow down our test suite.

TestRun

test.py
from typing import AsyncIterator

import httpx
import pytest

from service_a import app, service_b_client
from service_b import app as app_b


async def service_b_client_override() -> AsyncIterator[httpx.AsyncClient]:
    async with httpx.AsyncClient(app=app_b, base_url="http://test/b") as client:
        yield client


@pytest.fixture(name="client")
async def testclient():
    app.dependency_overrides[service_b_client] = service_b_client_override
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest.mark.anyio
async def test_call_b(client: httpx.AsyncClient) -> None:
    response = await client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

See more on the Starlette documentation.

Install the dependencies:

python -m pip install pytest httpx trio

Then open the terminal and run:

python -m pytest test.py

Good! Now are able to avoid a lot of network calls, and speed up our test suite.

November 3, 2022
in Testing
5 min read

Contract Testing with HTTPX

Today, we are going to talk about how to achieve contract testing with HTTPX.

What is contract testing?

Contract testing is a methodology for ensuring that two separate systems (such as two microservices) are compatible and are able to communicate with one other. It captures the interactions that are exchanged between each service, storing them in a contract, which can then be used to verify that both parties adhere to it. - Matt Fellows

What is HTTPX?

HTTPX is a fully featured HTTP client for Python 3, which provides sync and async APIs, and support for both HTTP/1.1 and HTTP/2.

How to do contract testing with HTTPX?

Well, to be completely transparent, I'm not sure if what you are about to read classifies as contract testing.

The problem we'll be trying to solve is the following:

Consider we have multiples services running, and they depend on each other. We want to make sure that a service is not able able to break another one.

To achieve this your first thought would be "let's write end to end tests", but that will slow things down, as each service needs to be up to run the tests, and given that, the setup needed is a bit more complex.

Check this blog post (which I didn't read, but looks good) for more information about E2E testing vs Contract Testing.

The solution

Let's assume we have two services. For obvious reasons, those services are FastAPI based.

Note

This can be achieved with any web framework. What matters here is that you should be using httpx.

Service AService BRun

service_a.py
import httpx
from fastapi import APIRouter, FastAPI

router = APIRouter(prefix="/a")


@router.get("/")
def get_a():
    return {"a": "a"}


@router.get("/call_b")
async def call_b():
    async with httpx.AsyncClient() as client:
        response = await client.get("http://localhost:8000/b/")
        return response.json()


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8001)  # (1)!

The port is 8001, not 8000, to avoid conflicts with the other service.

service_b.py
from fastapi import APIRouter, FastAPI

router = APIRouter(prefix="/b")


@router.get("/")
def get_b():
    return {"b": "b"}


app = FastAPI()
app.include_router(router)

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, port=8000)

Install the dependencies:

python -m pip install uvicorn fastapi httpx

Then open the terminal and run:

python service_a.py

Then open another terminal, and run:

python service_b.py

Cool. Now, let's call the /a/call_b endpoint:

http :8001/a/call_b # (1)!

The HTTP client used is called HTTPie, but you can use curl, or just go to the browser, and access http://localhost:8001/a/call_b.

As we see, the response is:

{
    "b": "b"
}

Now, if we want to create a test, we can do something like:

TestClientAsyncClientRun

test_service_a.py
import pytest
from fastapi.testclient import TestClient

from service_a import app


@pytest.fixture()
def client():
    return TestClient(app)


def test_call_b(client: TestClient) -> None:
    response = client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

See more on the Starlette documentation.

test_service_a.py
import httpx
import pytest
import pytest_asyncio

from service_a import app


@pytest_asyncio.fixture(name="client")
async def testclient():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest.mark.asyncio
async def test_call_b(client: httpx.AsyncClient) -> None:
    response = await client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

See more on the Starlette documentation.

Install the dependencies:

python -m pip install pytest pytest-asyncio httpx requests

Then open the terminal and run:

python -m pytest test_service_a.py

That works perfectly, right?

Well, yes. But remember what I said in the beginning?

To achieve this your first thought would be "let's write end to end tests", but that will slow things down, as each service needs to be up to run the tests, and given that, the setup needed is a bit more complex.

So, what if we want to run the tests without having to run the services?

Patch the HTTP client

We can patch the HTTPX client to make it call the service B, without actually running the service B.

To achieve that, we'll be using RESPX: a simple library, yet powerful, for mocking out the HTTPX and HTTPCore libraries.

It's easy, let me show you! We just need to add a single fixture on the test_service_a.py file:

TestClientAsyncClientRun

test_service_a.py
import pytest
import pytest_asyncio
import respx
from fastapi.testclient import TestClient

from service_a import app
from service_b import app as app_b


@pytest.fixture()
def client():
    return TestClient(app)


@pytest_asyncio.fixture()
async def create_contracts():
    async with respx.mock:  # (1)!
        respx.route(host="localhost", port=8000).mock(
            side_effect=respx.ASGIHandler(app_b)
        )
        yield


def test_call_b(client: TestClient) -> None:
    response = client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

The respx.mock context manager is used to mock the HTTPX client.

Read more about it on the RESPX documentation.

test_service_a.py
import httpx
import pytest
import pytest_asyncio
import respx

from service_a import app
from service_b import app as app_b


@pytest_asyncio.fixture(name="client")
async def testclient():
    async with httpx.AsyncClient(app=app, base_url="http://test") as client:
        yield client


@pytest_asyncio.fixture()
async def create_contracts():
    async with respx.mock:  # (1)!
        respx.route(host="localhost", port=8000).mock(
            side_effect=respx.ASGIHandler(app_b)
        )
        yield


@pytest.mark.asyncio
async def test_call_b(client: httpx.AsyncClient) -> None:
    response = await client.get("/a/call_b")
    assert response.status_code == 200
    assert response.json() == {"b": "b"}

The respx.mock context manager is used to mock the HTTPX client.

Read more about it on the RESPX documentation.

Install the dependencies:

python -m pip install pytest pytest-asyncio httpx respx requests

Then open the terminal and run:

python -m pytest test_service_a.py

Nice! We did it!

Now, we can run the tests without having to run the services.

If you are a curious person, feel free to compare the tests with the time command:

time python -m pytest test_service_a.py

Be surprised.

Info

You can also read the continuation of this article here.

September 20, 2022
in FastAPI
3 min read

FastAPI Escape Character

Today, we'll talk about a small feature of FastAPI that might be useful for you: the escape character.

What is the escape character?

The escape character \f is a character that can be used to tell to FastAPI to truncate what should go to the endpoint description on the OpenAPI.

Let's see it in practice. Consider we have the following endpoint:

CodeRun

main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def home():
    """This is home.
    \f
    This is not on the OpenAPI.
    """

Install the dependencies:

python -m pip install uvicorn fastapi

Then run uvicorn:

uvicorn main:app

When we call the /openapi.json endpoint:

http GET :8000/openapi.json  # (1)!

The HTTP client used is called HTTPie, but you can use curl, or just go to the browser, and access http://localhost:8000/openapi.json.

You'll see the following OpenAPI JSON on the response:

{
    "info": {
        "title": "FastAPI",
        "version": "0.1.0"
    },
    "openapi": "3.0.2",
    "paths": {
        "/": {
            "get": {
                "description": "This is home.",
                "operationId": "home__get",
                "responses": {
                    "200": {
                        "content": {
                            "application/json": {
                                "schema": {}
                            }
                        },
                        "description": "Successful Response"
                    }
                },
                "summary": "Home"
            }
        }
    }
}

Observe the "description" field does not contain the "This is not on OpenAPI" part of the docstring. The reason is the escape character we used. Everything after the \f will not appear on that field.

This feature may be useful if you are using a docstring linter tool, like darglint.

What's new?

If you are a FastAPI veteran (), you are probably familiar with the above. What you probably don't know, is that now (since FastAPI 0.82.0) it's possible to use it on the Pydantic models you use on your FastAPI application.

Let's see another example.

CodeRun

As most of my examples, we'll use potatoes:

main.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class PotatoOutput(BaseModel):
    """Super potato.
    \f
    This is not on the OpenAPI.
    """

@app.get("/", response_model=PotatoOutput)
def get_potato():
    ...

Install the dependencies:

python -m pip install uvicorn fastapi

Then run uvicorn:

uvicorn main:app

When we call /openapi.json, as we did above, we'll get the following OpenAPI JSON as response:

{
    "components": {
        "schemas": {
            "PotatoOutput": {
                "description": "Super potato.\n",
                "properties": {},
                "title": "PotatoOutput",
                "type": "object"
            }
        }
    },
    "info": {
        "title": "FastAPI",
        "version": "0.1.0"
    },
    "openapi": "3.0.2",
    "paths": {
        "/": {
            "get": {
                "operationId": "get_potato__get",
                "responses": {
                    "200": {
                        "content": {
                            "application/json": {
                                "schema": {
                                    "$ref": "#/components/schemas/PotatoOutput"
                                }
                            }
                        },
                        "description": "Successful Response"
                    }
                },
                "summary": "Get Potato"
            }
        }
    }
}

Tip

We can also use jq to get the part of the JSON that we are interested.

CallOutput

http GET :8000/openapi.json | jq .components.schemas

{
    "PotatoOutput": {
        "title": "PotatoOutput",
        "type": "object",
        "properties": {},
        "description": "Super potato.\n"
    }
}

As we can see, the description of PotatoOutput doesn't contain the "This is not on the OpenAPI." part as well.

Yey! Now you can use those docstring linter tools as you want with FastAPI!

Thanks for reading this blog post!

If you have any suggestions on what I can write about, please feel free to suggest below.

September 13, 2022
in Testing
5 min read

FastAPI's Test Client

This is my first blog post!

Please enjoy, and let me know if you have any feedback.

Abstract

If you are new to FastAPI, you might benefit from reading the following:

If you already know stuff about FastAPI, you might jump to:

The Future of the TestClient

Today, we'll talk about the main tool for testing FastAPI applications: the TestClient.

`TestClient` origin and features

The TestClient is a feature from Starlette (one of the two main dependencies of FastAPI). On which, FastAPI only does a reimport on the testclient module, as we can see here.

We can use the TestClient to test our WebSocket and HTTP endpoints.

The `TestClient` weird behavior

Although documented on both FastAPI and Starlette's documentation, most of the people are not aware of the TestClient's behavior when it comes to events. To put it simple, there are two ways of creating a TestClient object, and in one of those ways, the events are not executed.

Let's see the behavior with the following FastAPI application:

main.py
from fastapi import FastAPI

app = FastAPI()
started = False

@app.on_event("startup") # (1)!
def startup():
    global started
    started = True

@app.get("/")
def home():
    if started:
        return {"message": "STARTED"}
    else:
        return {"message": "NOT STARTED"}

There are only two events available: startup and shutdown.

Read more about it on the ASGI documentation.

As you can see, there's a single endpoint, which gives us a different message depending on the value of the started variable. The started variable is set to True on the startup event.

Now, let's test it with the TestClient:

CodeRun

test.py
from fastapi.testclient import TestClient

from main import app

def test_home():
    client = TestClient(app)
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "NOT STARTED"}

Install the dependencies:

python -m pip install "fastapi[all]" pytest

Then run pytest:

pytest test.py

As you can see, the above test passes. Which means the startup event was not triggered.

On the other hand, if we run the following test, we'll get a different result:

CodeRun

test.py
from fastapi.testclient import TestClient

from main import app

def test_home():
    with TestClient(app):
        response = client.get("/")
        assert response.status_code == 200
        assert response.json() == {"message": "STARTED"}

Install the dependencies:

python -m pip install "fastapi[all]" pytest

Then run pytest:

pytest test.py

When used as context manager, the TestClient will trigger the startup event.

The Future of the `TestClient`

By the moment I'm writing this blog, the latest FastAPI version is 0.83.0 with Starlette pinned on 0.19.1. Starlette is already on version 0.20.3, and the next release will change the internals of the TestClient. To be more specific, the HTTP client will be changed from requests to httpx.

As there are some differences between the two clients, the TestClient will reflect the same differences.

This change will be in Starlette on version 0.21.0, and I'm unsure when it will land on FastAPI.

Let's see the changes you should be aware:

allow_redirects will be now called follow_redirects.
cookies parameter will be deprecated under method calls (it should be used on the client instantiation).
data parameter will be called content when sending bytes or text.
content_type will default to "text/plain" when sending file instead of empty string.
The HTTP methods DELETE, GET, HEAD and OPTIONS will not accept content, data, json and files parameters.

data parameter doesn't accept list of tuples, instead it should be a dictionary.

❌ List of Tuples✅ Dictionary

client.post(..., data=[("key1", "1"), ("key1", "2"), ("key2", "3")])

client.post(..., data={"key1": ["1", "2"], "key2": "3"})

Those changes will likely impact your test suite. Having this in mind, I've created a codemod that will help you to migrate your tests: bump-testclient.

Here is the list of what the codemod will do:

Replace allow_redirects with follow_redirects.
Replace data with content when sending bytes or text.
Replace client.<method>(..., <parameter>=...) by client.request("<method>", ..., <parameter>=...) when parameter is either content, data, json or files.

In case you want to read more about the differences between the underneath clients, you can check the httpx documentation.

Thanks for reading till here!