DEV Community: Rafael de Oliveira Marques

Understanding Uvicorn: The basics

Rafael de Oliveira Marques — Thu, 13 Mar 2025 12:26:21 +0000

Last year I wrote a series of articles explaining how FastAPI works. It was a rewarding experience, since I was able to both share my knowledge and learn more about it.

So I've decided to create another series of articles, and this time talking about another piece of FastAPI development: ASGI servers, and mainly focusing on 🦄 uvicorn due to the evident synergy between both projects 😎.

🤔 So, what is uvicorn?

As the project says in a very direct and clear way:

Uvicorn is an ASGI web server implementation for Python.

So, following the same approach as the FastAPI articles, we'll try to understand some concepts and libs before moving on.

As the creators of the 🦄 uvicorn says, it is an ASGI web server implementation.

So, let's take a small look at what is a web server.

💻 What is a web server?

Ade56facc, CC BY-SA 4.0 , via Wikimedia Commons

From the name, we can try to infer that this is a server used on the web, right? But let's look at some definitions to understand it better.

At Mozilla Developer Network, we can find the definition in the What is a web server? document:

The term web server can refer to hardware or software, or both of them working together...
...On the software side, a web server includes several parts that control how web users access hosted files. At a minimum, this is an HTTP server. An HTTP server is software that understands URLs (web addresses) and HTTP (the protocol your browser uses to view webpages). An HTTP server can be accessed through the domain names of the websites it stores, and it delivers the content of these hosted websites to the end user's device.

So, going back to 🦄 uvicorn, as it is an ASGI web server, we can begin to imagine that maybe it is a server that handles HTTP and implements the ASGI specification, right?

📜 And what is HTTP?

HTTP stands for HyperText Transfer Protocol, and as the name says, it's a Protocol to transfer HyperText. Wow, mind blowing, right? 🤯

Since a Protocol is a set of rules for comunication, we can now understand that HTTP is a set of rules of communication for transfering HyperText. And since we can say in a summarized way that HyperText is a text that references hyperlinks, we can now understand that:

Uvicorn is a software that enables two or more computers to communicate and transfer text throughout a set a rules defined in the HTTP Protocol.

⚠️ I know, I know. Uvicorn supports WebSockets too. But let's stick with the basics for now. 😊

💾 Creating your first socket

We just saw that 🦄 uvicorn is a software that enables communication between computers using some standards, etc.

The way it enables the communication is throughout sockets, since uvicorn understands HTTP, and HTTP uses TCP/IP, which commonly uses sockets for the communication.

Let's start creating our first socket that will simply echo the received message:

import socket

print("Creating server")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server:
    server.bind((socket.gethostname(), 12345))
    print(f"Listening at {socket.gethostname()}:12345")

    server.listen()

    print("Waiting for connection")
    connection, adress = server.accept()

    while True:
        data = connection.recv(1024)
        print(f"Received: {data.decode()}")
        connection.sendall(data)

What we've just created is a server that will keep echoing forever the messages that were received:

Creating server
Listening at ceb10n:12345
Waiting for connection
Received: My first socket connection :D

Now, we need to create a client socket to connect to our server and see the connection in action:

import socket

print("Creating client")
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    print(f"Connecting to {socket.gethostname()}:12345")
    s.connect((socket.gethostname(), 12345))
    while True:
        s.sendall(b"My first socket connection :D")
        data = s.recv(1024)
        print(f"Received {data!r}")

And the result will be:

Creating client
Connecting to ceb10n:12345
Received b'My first socket connection :D'

But what we just did right here?

We created a socket specifying the parameters socket.AF_INET and socket.SOCK_STREAM.

socket.AF_INET: It's the address family of our socket. In this case, our socket will use the IPv4.
socket.SOCK_STREAM: It's the socket type. This way, our socket will use the TCP protocol (the first step for creating our HTTP server)

The next part of our server is binding the our socket with a host and port.

Then we need to start accepting a connection. We do this by enabling the socket to accept connections with listen. After our socket is enabled, we can use the accept method to wait for a connection.

After that, we start a loop to keep receiving chunks of data forever. The recv method receives the 1024 parameter, specifying that we want to receive blocks of 1024 bytes of data. You can test it by changing 1024 to only one or two. If you do that, you'll start seeing the following:

Creating server
Listening at ceb10n:12345
Waiting for connection
Received: My

Notice that if the message is bigger than what we've specified there, we'll lose all the remaining bytes of the message.

And probably that's the most basic socket server you can create. On the client side, it's almost identical, except we'll be sending data with sendall, that receives the message we want to send as a parameter.

Now we start figuring out the amazing work that the uvicorn team (and the teams from the other dependencies) have done! Just imagine starting from this simple socket server and transforming it in the amazing server that is uvicorn! 🤯

I would suggest you all to consider 💸 sponsoring the project and the main maintainer, 🇧🇷 Marcelo Trylesinski! Just visit it's 🦄 official repository and start supporting them!

Well, I think that's all for this article. Stay tuned for the next ones 😎.

FastAPI: How to use Pydantic to declare Query Parameters

Rafael de Oliveira Marques — Wed, 09 Oct 2024 16:49:45 +0000

It came out about three weeks ago one of the most expected features of FastAPI. At least when we're talking about Pydantic Models + FastAPI.

Yes, I'm talking about the ability to use Pydantic Models to map your query parameters.

So in this post, I'll try to show you all you 👍 can and 👎 can't do about this subject 🙂:

💻 Mapping Query Parameters

The first thing you need to do to start mapping your query parameters with Pydantic is making sure you are using FastAPI version 0.115.0 and above.

After this, you can always go to FastAPI docs to check what is already available. Sebastián and the team members make a really, really good work on keeping do docs updated and informative ✨.

📜 A little bit of History

Let's start with some examples on how we used to map Query Parameters in FastAPI. 🤓

The simplest way to do it would be:

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def search(
    limit: int | None = 10,
    skip: int | None = 1,
    filter: str | None = None
):
    return {
        "limit": limit,
        "skip": skip,
        "filter": filter
    }

And now you can simply call:

GET http://localhost:8000/?limit=42&skip=12&filter=banana

But if we identified that this Query Parameters would be used in other routes, we would isolate it with something like:

from typing import Any
from fastapi import Depends, FastAPI, Query

app = FastAPI()

async def pagination_query_string(
    limit: int | None = Query(10, ge=5, le=100),
    skip: int | None = Query(1, ge=1),
    filter: str | None = Query(None)
) -> dict[str, Any]:
    return {
        "limit": limit,
        "skip": skip,
        "filter": filter
    }

@app.get("/")
async def search(q: dict[str, Any] = Depends(pagination_query_string)):
    return q

Or since we're using Pydantic to map our models, with just a little refactoring we would get:

from fastapi import Depends, FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class PaginationQueryString(BaseModel):
    limit: int | None = 10
    skip: int | None = 1
    filter: str | None = None

async def pagination_query_string(
    limit: int | None = Query(10, ge=5, le=100),
    skip: int | None = Query(1, ge=1),
    filter: str | None = Query(None)
) -> PaginationQueryString:
    return PaginationQueryString(
        limit=limit,
        skip=skip,
        filter=filter
    )

@app.get("/")
async def search(q: PaginationQueryString = Depends(pagination_query_string)):
    return q

⌨️ Using Pydantic to map the Query Strings

Now, if we want to get our query string, we don't need to create a function and then add it as a dependency. We can simply tell FastAPI that we want an object of type PaginationQueryString and that it's a query string:

from typing import Annotated
from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class PaginationQueryString(BaseModel):
    limit: int | None = 10
    skip: int | None = 1
    filter: str | None = None

@app.get("/")
async def search(q: Annotated[PaginationQueryString, Query()]):
    return q

Easy, right? 😄

❌ Forbid extra Query Parameters

To forbid extra parameters in our query strings we need to configure our Pydantic model to do so.

We'll simply add a model_config with the option extra="forbid":

from typing import Annotated
from fastapi import FastAPI, Query
from pydantic import BaseModel, ConfigDict

app = FastAPI()

class PaginationQueryString(BaseModel):
    model_config = ConfigDict(extra="forbid")

    limit: int | None = 10
    skip: int | None = 1

@app.get("/")
async def search(q: Annotated[PaginationQueryString, Query()]):
    return q

Note that we removed the filter parameter in our model. Now if we call:

GET http://localhost:8000/?limit=42&skip=12&filter=banana

We'll get an error informing us that extra inputs are not allowed:

{
    "detail": [
        {
            "type": "extra_forbidden",
            "loc": [
                "query",
                "filter"
            ],
            "msg": "Extra inputs are not permitted",
            "input": "banana"
        }
    ]
}

⚠️ What are the limitations?

At least at version 0.115.0, it don't work very well with nested models.

Let's try something like:

from typing import Annotated
from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI()

class Filter(BaseModel):
    name: str | None = None
    age: int | None = None
    nickname: str | None = None

class PaginationQueryString(BaseModel):
    limit: int | None = 10
    skip: int | None = 1
    filter: Filter | None = None

@app.get("/")
async def search(q: Annotated[PaginationQueryString, Query()]):
    return q

If we call it like before:

GET http://localhost:8000/?limit=42&skip=12&filter=chocolate

We'll get an error telling us that filter is an object:

{
    "detail": [
        {
            "type": "model_attributes_type",
            "loc": [
                "query",
                "filter"
            ],
            "msg": "Input should be a valid dictionary or object to extract fields from",
            "input": "chocolate"
        }
    ]
}

At least right now, it's absolutely right! We changed our filter to be a Pydantic model, not a string. But if we try to convert it to a dictionary:

http://localhost:8000/?limit=42&skip=12&filter={%22name%22:%20%22Rafael%22,%20%22age%22:%2038,%20%22nickname%22:%20%22ceb10n%22}

FastAPI will tell us that filter needs to be a valid dictionary 😔:

{
    "detail": [
        {
            "type": "model_attributes_type",
            "loc": [
                "query",
                "filter"
            ],
            "msg": "Input should be a valid dictionary or object to extract fields from",
            "input": "{\"name\": \"Rafael\", \"age\": 38, \"nickname\": \"ceb10n\"}"
        }
    ]
}

It's happening this because FastAPI will rely on Starlette's QueryParams, that will give a string to FastAPI, not a dict. And at least in version 0.115.0, this will give you an error.

⁉️ So, when do I use Pydantic models with my Query Parameters?

It's quite simple:

✅ You have simple query strings that don't need any elaborate fancy nested objects? Use it! 👍

❌ You created a complex nested query string? Don't use it yet 👎. (And maybe you should try to rethink your query strings. 😇 The simpler, the better 😉)

Helping FastAPI: How to contribute to docs translations

Rafael de Oliveira Marques — Wed, 04 Sep 2024 11:27:48 +0000

One of the great features of FastAPI is its great documentation 👏. But wouldn't it be better if more people around the world had access to this documentation? ❤️

Sometimes we take for granted that all people can read the docs we provide, but it's not necessarily how it works to people around the world.

📝 Why add translations?

📖 There is already a really good documentation written in english in FastAPI website, so why help translate to other languages?

Making a fast google search, you can see that only 17% of the 🌎 world population speaks english. You can check this wikipedia post: List of countries by English-speaking population to check the percentage of english speakers in your country.

For example, I'm a brazilian living in Brazil. And here, only 5% of the population have some level of english. This represents a very small portion of the population that can follow documentation written in English.

And continuing with the portuguese, it's not only Portugal and Brazil that speaks this language. There's also Angola, Mozambique, Cape Verde and many other countries. You can see the full list here.

Do you have any idea on how many people you can help when you translate the docs from your favorite programming language or framework? It's mindblowing thinking in the number of people who benefit from it.

Additionally, helping with translations is a very practical way to understand the way the project works, the workflow and approval that maintainers follow, etc.

✏️ How to create your first translation

FastAPI docs has a page dedicated on how to contribute to the project, including a section for the documentations and translations.

So let's take a look step by step on how you can setup your local environment and start creating translations to a language other then english!

🍴 Forking FastAPI's repository

The first thing that you want to do is fork the FastAPI repo. Github has a great doc explaining how you can fork a repository. But basically, what you'll need to do is browse the repo you want, in this case FastAPI's repo and click Fork.

And that's it. Know you have your own copy of the repository. Now if you browse your own repositories, you'll see the forked repo right there:

🗂️ FastAPI's documentation structure

The FastAPI's documentation lives inside the folder docs, in the root of the repository. And all the source code from the docs lives inside docs_src folder.

As you can see, inside docs folder, there are all the current languages that has translations. It uses the ISO 693-1 two letter code for each language.

Each language folder will follow the same structure:

The en folder will have the complete documentation, but you'll notice that in other languages, despite having the same structure, not all files are present. And that's because not all files are translated to all languages (yet 😇).

💡 So now you know the first way to find what you can translate: There is a missing file in your language? That's the one you can start translating!

☹️ Missing languages

Not all languages has translations yet. For example, if you look for the 🇦🇲 Armenian code (hy) in the docs, you'll notice the it doesn't exist yet:

In these cases, FastAPI docs has a really good explanation on how you can add translations to new languages.

As you can see from the docs, FastAPI has a neat script to initialize a new language translation:



python ./scripts/docs.py new-lang hy

Now that the script added the folder and files, you can follow the process of adding translations to existing languages.

🗺️ Translating existing languages

Now that you already forked the FastAPI's repository and learned how the add a missing language (if its your case), let's see the whole process of translating a file of the docs.

📋 Ways to verify what doc to translate

There are a few ways you can easily find what doc you can translate.

1️⃣ Navigating through the docs

A simple and easy way is: Read the docs in your desired language. Just go to the docs at fastapi.tiangolo.com and select the desired language:

Once you reach a doc that has no translations, you'll see a warning telling you that the doc has not been yet translated:

Now you can go to the source code, find the doc file and create a copy at your desired language folder. The folder structure of the docs are pretty simple to understand.

In the example above, the breadcrumbs say that we are at: FastAPI (The root) - Learn - Advanced. So we can infer that the document lives somewhere in docs/en/docs. Probably in some folder named learn or advanced.

If we go to the source code, will see that the folder advanced really exists, and the file custom-response.md also exists.

An easier way to find the file is get the last part of the url and find for it in your editor. For example, in VSCode you can use ctrl + e and enter that name:

2️⃣ Looking in source code

Another way you can find files that has no translations it to open the source code with your editor. Then you'll expand the desired language and the english language.

You will probably notice that the English folder has more files than the folder for your desired language. Now you can pick the missing file, create a copy of it at the language folder and translate it.

3️⃣ Using FastAPI Translations Management package

I created a 📦 package to help with FastAPI translations named fastapi translations management.

It's basically a lib that will look at all doc files and check the last commit date. This will give you a list of files that has not been translated yet and the files that are outdated.

To use it, you can install it with pip:



pip install -U fastapi-translations

And then use it at the root folder of your forked FastAPI repository:



fastapi_translations -l {two-letter-code-for-language}

This will give you a brief summary of missing translations and outdated translations:

You can also generate a csv file with all files that are outdated and missing with -c:



fastapi_translations -l pt -c

📋 Things to know before start your first translation

💬 Discussions

If you want to translate to a language that already exists, probably it has a Topic created under Discussions. You can search your language in github.com/fastapi/fastapi/discussions.

At portuguese discussions, we have a pattern to always post the file we are translating, and after we finish, we edit it to add the PR link:

🔍 Reviews

All pull requests for translations must have at least two approvals from a person who speaks that language.

For example, if you create a translation for Japanese. Two people that speaks Japanese must review it before some mantainer approves the PR.

👨🏾‍⚖️ Another rules

There are some other rules that apply when we are translating some docs.

✅ Don't translate the code in docs_src;
✅ Only translate the markdown files;
✅ Inside code blocks, only translate the # comments;

You can check all the rules in FastAPI docs for tips and guidelines for translations.

✨ Creating your first translation

Now that we know almost everything that is to be know about translating FastAPI docs, let's get started and translate a new doc.

⚙️ Update your FastAPI fork

Whenever you start a new translation, you need to update your forked repository to make sure everything is updated ✔️.

The easiest way to do this is to navigate to your repository at github and click in sync fork -> update branch.

Now you can update your local repository with all changes from the main repo.

🔎 Find the doc to translate

Now that our local repository is updated. Let's find some missing translation.

We can see that under docs/pt/docs/advanced, the 📁 folder security is missing. So let's translate the index.md for the advanced security topic.

🙋🏿‍♀️ Notifying about translation in progress

Now that we picked a file to translate, let's tell everyone that in the 💬 Discussion for Portuguese translations that we are working on it:

🛠️ Creating the translation

Not let's create a branch for the translation:



git checkout -b features/pt-advanced-security-index

Since we working on our local forked repository, we don't necessarily need to create a specific branch. But I think it's a good thing to do. And working this way, we can start another work while people are reviewing our PR.

Now we can create both the missing folder 📁, and the missing file 📃 under docs/pt/docs/advanced.

When I'm translating some file, I like to split the editor with the file that I'm working on, and the original file in english. But feel free to work the way is best for you.

Now that we finished our work translating the file, we can commit it:



git add docs/pt/docs/advanced/security/index.md
git commit -m "Add translation to docs/pt/docs/advanced/security/index.md"
git push origin features/pt-advanced-security-index

👀 Previewing the translation

Now that we finished the translation, we can see how it will look like on the official docs.

You can type in your terminal 👨🏼‍💻 (remember to install all deps):



python scripts/docs.py live pt

And you'll be able to check the result:

💻 Creating the Pull Request

Remember that we are working on our fork. Now that we commited to our repository, we need to send it to the FastAPI repository. Luckily, this is very easy to do.

If you go to the FastAPI repository, github will warn you that you pushed to your fork, and now you can create a PR to merge it:

We can click on compare & pull request and create the PR following the pattern for the title:

🌐 Add Portuguese translation for path/of/file.md

Now we can wait for all the checks to run (they must pass). And someone from the FastAPI team will add the necessary tags.

And of course, we need to update our post at the discussions to inform that we finished the translation:

And after everything goes well, you'll get a message telling you that your PR was approved ✨:

💣 Dealing with problems

I didn't anticipate this when I started writing this article. A problem related with github actions and upload-artifact started happening and the checks from my PR failed 💥.

This was a really nice thing to happen to demonstrate how we can deal with situations that our PR has some problems.

When I saw the failing checks, I tried to see if it was related with my PR directly. I saw it was not related, and then I marked Alejandra, who is a very helpful member of the FastAPI team. Sofie, who is also a member of the team mentioned the issue related with the problem right away.

As you can see, FastAPI has a really nice and helpful team and they always do their best to help you:

So if you need some help, try to reach one the them. Just be polite and patiente that they will help you ❤️!

🎁 Benefits of translating docs

There are several benefits helping with the translations 🎉.

To me, the most important one is to help people who have difficulty reading documentation in english.

They can be 👩🏽‍🎓 students trying to learn a new language or framework, or even a 👩‍💻 professional who has not yet had the opportunity to learn english.

In addition to helping people, you can also 📖 learn new topics, find out about that detail that you used but didn't quite understand why, etc.

Also, helping translate docs requires you to review the original documentation. This may result in you finding improvements, topics that could be better explained, etc.

So, my advice is: do you have a language or framework that you really like and would like to start helping? Start with the documentation 📚!

Pydantic Settings + AWS the easy way

Rafael de Oliveira Marques — Sun, 28 Jul 2024 18:41:05 +0000

Pydantic Settings is a python library that extends 🚀 Pydantic for dealing with settings management.

If you've never heard of pydantic, please, check it's docs page to see how easy and powerfull it is. In my opinion, it is one of the best python libraries I have come across in the last few months/years. Congrats Samuel Colvin!

⚙️ Using Pydantic Settings

To use pydantic-settings, we first need to install it:



pip install pydantic
pip install pydantic-settings

Now let's imagine that we have a FastAPI microservice that needs to load some configuration.

Let's say we depend on some information to work properly:

Environment that is available in the environment variables
Database host that is available in an AWS Parameter Store
Database authentication info that is available in an AWS Secrets Manager

What we would need to do is something like:



import json

import boto3

from pydantic_settings import BaseSettings, SettingsConfigDict


class AppSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="APP_")

    environment: str  # will be loaded by environment variable APP_ENVIRONMENT
    host: str  # need to pass it when creating AppSettings :(
    username: str  # need to pass it when creating AppSettings :(
    password: str  # need to pass it when creating AppSettings :(


ssm_client = boto3.client("boto3")
secrets_client = boto3.client("secretsmanager")

# need to get the secret before creating our settings object
secret_response = secrets_client.get_secret_value(SecretId="my/secrets")
secret_string = secret_response.get("SecretString")
secret = json.loads(secret_string)

# need to get the parameter store before creating our settings object
parameter = ssm_client.get_parameter(Name="/my/ssm/parameter")
host = parameter.get("Parameter").get("Value")


settings = AppSettings(host=host, **secret)

Don't get me wrong, it's still great. But it would be better if we had a pydantic settings source to do that work for us!

Obs: You can check pydantic-settings 📖 docs to see all settings sources that it offers.

☁️ Pydantic Settings AWS

I recently started creating a pydantic-settings extension to deal with settings that lives at AWS. It called Pydantic Settings AWS and you can see the source code at github.com/ceb10n/pydantic-settings-aws

🏃 Gettings started

If you still don't have boto3 or pydantic-settings installed:



pip install boto3
pip install pydantic-settings

And then you can install pydantic-settings-aws



pip install pydantic-settings-aws

Now let's get back to our example.

We have a basic settings class that needs to load data from Environmet variables, parameter store and secrets manager.

pydantic-settings-aws offers a class called AWSBaseSettings, that gives you the ability to deal with all that data sources.



from typing import Annotated

from pydantic_settings import SettingsConfigDict
from pydantic_settings_aws import AWSBaseSettings


class AppSettings(AWSBaseSettings):
    model_config = SettingsConfigDict(
        secrets_name="my/secrets",
        env_prefix="APP_"
    )

    environment: str
    host: Annotated[str, {"service": "ssm", "ssm": "/my/ssm/parameter"}]
    username: Annotated[str, {"service": "secrets"}]
    password: Annotated[str, {"service": "secrets"}]


settings = AppSettings()

With AWSBaseSettings, all you need to do is using typing.Annotated to inform what service you want to use, and other things, like the secret and parameter store names.

🙊 Settings with Secrets Manager only

If you simply want to use Secrets Manager, you can use the SecretsManagerBaseSettings.

Since we are using only Secrets Manager as a data source, we don't need to specify anything, except the secrets name:



from pydantic_settings import SettingsConfigDict
from pydantic_settings_aws import SecretsManagerBaseSettings


class AppSettings(SecretsManagerBaseSettings):
    model_config = SettingsConfigDict(
        secrets_name="my/secrets"
    )

    username: str
    password: str

settings = AppSettings()
# username='my-username' password='my-super-secret-password'

🔒 Settings with Parameter Store only

Let's say you use Parameter Store to save a lot of you apps configuration, like queue names, urls, etc.

You can leverage the ParameterStoreBaseSettings class to load all data from different parameter stores.




from pydantic_settings_aws import ParameterStoreBaseSettings


class AppSettings(ParameterStoreBaseSettings):

    dev_base_endpoint: str
    database_host: str
    database_port: str

And thats it. ParameterStoreBaseSettings will try to create a boto3 client with your local configuration, and then will try to get the value from a parameter store with the same name as your field.

But it's not uncommon for the names of our parameter stores to not match the names of our variables. In theses cases, you can use typing.Annotated:



from typing import Annotated

from pydantic_settings_aws import ParameterStoreBaseSettings

class AppSettings(ParameterStoreBaseSettings):

<span class="n">dev_base_endpoint</span><span class="p">:</span> <span class="n">Annotated</span><span class="p">[</span><span class="nb">str</span><span class="p">,</span> <span class="sh">"</span><span class="s">/payments/endpoint</span><span class="sh">"</span><span class="p">]</span>
<span class="n">database_host</span><span class="p">:</span> <span class="n">Annotated</span><span class="p">[</span><span class="nb">str</span><span class="p">,</span> <span class="sh">"</span><span class="s">/databases/mongodb/payments/dbhost</span><span class="sh">"</span><span class="p">]</span>
<span class="n">database_port</span><span class="p">:</span> <span class="n">Annotated</span><span class="p">[</span><span class="nb">str</span><span class="p">,</span> <span class="sh">"</span><span class="s">/databases/mongodb/payments/dbport</span><span class="sh">"</span><span class="p">]</span>

📜 Docs and source code

You can take a look at pydantic-settings-aws documentations at [ceb10n.github.io/pydantic-settings-aws](https://ceb10n.github.io/pydantic-settings-aws/].

The project is hosted at github: github.com/ceb10n/pydantic-settings-aws. It's still an ongoing project. Feel free to open an issue, make a PR, etc. 🤗

Understanding FastAPI: How OpenAPI works

Rafael de Oliveira Marques — Wed, 24 Jul 2024 11:50:05 +0000

We already saw what is ASGI, how starlette works and how FastAPI extends starlatte. Now it's time to see how FastAPI offers one of the greatest features (in my opinion): Out of the box API docs with Swagger and Redoc.

💭 What is OpenAPI?

If we go to the OpenAPI's repository, we'll see that:

The OpenAPI Specification (OAS) defines a standard, programming language-agnostic interface description for HTTP APIs. This allows both humans and computers to discover and understand the capabilities of a service without requiring access to source code, additional documentation, or inspection of network traffic. When properly defined via OpenAPI, a consumer can understand and interact with the remote service with a minimal amount of implementation logic. Similar to what interface descriptions have done for lower-level programming, the OpenAPI Specification removes guesswork in calling a service.

So the OpenAPI is simply a document that tells you how you can define and implement an API following a standard, making it easier for other developers to understand and consume.

If you have ever worked with any type of systems integration, you know how easy or extremely difficult your life can be, depending on the quality of the integration documentation.

And here is where FastAPI ✨ shines: It leverages Pydantic powerful data validation to offer out of the box JSON Schema and OpenAPI specs via Swagger and Redoc.

How FastAPI offers JSON Schema and OpenAPI?

FastAPI will give you for free OpenAPI docs with both Swagger and Redoc.

The JSON Schema is offered using Pydantic, and if you are using FastAPI, you probably already know it.

🚀 Pydantic and JSON Schema

See how easy is to generate a JSON Schema with Pydantic:



from pydantic import BaseModel, Field, TypeAdapter


class User(BaseModel):

    id: str = Field(
        ...,
        title="User ID",
        description="User's unique identification"
    )

    name: str = Field(
        ...,
        title="Name",
        description="User's name",
        min_length=5,
        max_length=50
    )


type_adapter = TypeAdapter(User)

Now if we call TypeAdapter's json_schema method, we'll get a valid JSON Schema compliant with OpenAPI Specification v3.1.0:



{
    "properties": {
        "id": {
            "description": "User's unique identification",
            "title": "User ID",
            "type": "string"
        },
        "name": {
            "description": "User's name",
            "maxLength": 50,
            "minLength": 5,
            "title": "Name",
            "type": "string"
        }
    },
    "required": [
        "id",
        "name"
    ],
    "title": "User",
    "type": "object"
}

⚙️ FastAPI documentation setup

When we create a FastAPI instance, it will setup all you need to create the docs.

The 🧙 magic will happen inside setup function, that is called inside __init__.

If we look at setup function, we'll see that FastAPI creates for us an openapi route unless we set the openapi_url=None.



if self.openapi_url:
    urls = (server_data.get("url") for server_data in self.servers)
    server_urls = {url for url in urls if url}

    async def openapi(req: Request) -> JSONResponse:
        root_path = req.scope.get("root_path", "").rstrip("/")
        if root_path not in server_urls:
            if root_path and self.root_path_in_servers:
                self.servers.insert(0, {"url": root_path})
                server_urls.add(root_path)
        return JSONResponse(self.openapi())

    self.add_route(self.openapi_url, openapi, include_in_schema=False)

FastAPI is setting up here the /openapi.json route, that will call the get_openapi function. It will return all the Schema from our project, or more precisely, an OpenAPI object defined inside openapi module.

So the order of events is:

When you create your app, FastAPI will add the routes:

/openapi.json
/docs
/redoc

if you don't remove it explicitly.

The /openapi.json route will return a JSONResponse containing all needed data from your routes when called.

The /docs and /redoc routes will basically return an HTML that loads the swagger and redoc respectively.

📑 Minimal example

If we create a minimal application:



from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def hello():
    return {"Hello": "world :D"}

We can see our swagger docs calling /docs:

And redoc docs calling /redoc:

📰 Setting up Redoc

We have seen that redoc will be available for free when you create a FastAPI project.

What is happening here is that FastAPI will define a redoc url by default:



if self.openapi_url and self.redoc_url:
    async def redoc_html(req: Request) -> HTMLResponse:
        root_path = req.scope.get("root_path", "").rstrip("/")
        openapi_url = root_path + self.openapi_url
        return get_redoc_html(
            openapi_url=openapi_url, title=f"{self.title} - ReDoc"
        )

    self.add_route(self.redoc_url, redoc_html, include_in_schema=False)

Since both openapi_url and redoc_url has default values of /openapi.json and /redoc respectively, redoc will be served at /redoc by default.

If you want to remove redoc from your project, you just need to set redoc_url=False:



app = FastAPI(redoc_url=None)

If you want to remove OpenAPI docs too, you just need to set openapi_url to None:



app = FastAPI(openapi_url=None)

📃 Setting up Swagger

Swagger will work the same way as redoc. The path /docs will be added automaticaly if don't pass any value to docs_url. If you inform a new path, swagger will be served in this new url:



if self.openapi_url and self.docs_url:

    async def swagger_ui_html(req: Request) -> HTMLResponse:
        root_path = req.scope.get("root_path", "").rstrip("/")
        openapi_url = root_path + self.openapi_url
        oauth2_redirect_url = self.swagger_ui_oauth2_redirect_url
        if oauth2_redirect_url:
            oauth2_redirect_url = root_path + oauth2_redirect_url
        return get_swagger_ui_html(
            openapi_url=openapi_url,
            title=f"{self.title} - Swagger UI",
            oauth2_redirect_url=oauth2_redirect_url,
            init_oauth=self.swagger_ui_init_oauth,
            swagger_ui_parameters=self.swagger_ui_parameters,
        )

    self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)

As we can see, the setup function will add the swagger ui url to our app unless we define docs_url=None.

👋 and that's all for today. See you in the next article 🤗

Understanding FastAPI: How FastAPI works

Rafael de Oliveira Marques — Sun, 30 Jun 2024 20:17:10 +0000

At this point we've seen how ASGI servers and our applications talk to each other and how Starllete, the foundation of FastAPI works.

Now it's time to take a closer look on how FastAPI extends Starllete.

FastAPI, a Starllete app

First of all, to understand how FastAPI works, the are two main sources of information:

So make sure you clone Sebastián's repository and start looking at it.

FastAPI's first entrypoint is FastAPI class, that lives under fastapi/applications.py.

Since we are studying an ASGI framework, we can expect that FastAPI is a callable that receives scope, receive and send, like any other ASGI app:

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    if self.root_path:
        scope["root_path"] = self.root_path
    await super().__call__(scope, receive, send)

We can see that not only FastAPI has a __call__ function as we expected, but it delegates to Starlette the request.

Difference between FastAPI and Starlette when initializing

If FastAPI extends Starlette, it will likely add some functionality during initialization.

When we look at FastAPI's __init__ function, we can see two main things:

It add routes to OpenAPI docs on setup function
It sets the Router to APIRouter

setup function will add one of the coolest features of FastAPI: It will add a free OpenAPI documentation to our project with Swagger and Redoc.

The APIRouter will be where all your path operations live. Either you add your route directly with your FastAPI app ou creating an APIRouter, all routes will be included in FastAPI's router.

In this post we'll take a better look at FastAPI's routers and routes. We'll leave OpenAPI to a next post.

Request lifecycle

Since FastAPI is a Starlette app with extra features, we can assume that a request lifecycle using FastAPI will be almost equal to Starlette's request lifecycle.

On the previous post, we talked about how a request will be handled. The chain of middlewares will be something like:

-> ServerErrorMiddleware
    -> Other Middlewares
        -> ExceptionMiddleware
            -> Router

When working with FastAPI we can see the it is overriding Starlette's Router with it's own APIRouter.

That said, when can see that FastAPI is still relying on Starlette's lifecycle, but it prefers to handle the requests its own way.

So with FastAPI, you'll have:

-> FastAPI App
  -> Starlette's App
    -> Starlette's ServerErrorMiddleware
      -> Starlette's ExceptionMiddleware
        -> FastAPI's APIRouter (and Router, since it don't override Router's __call__)

FastAPI routers and routes

When we are creating a FastAPI app, there are two main ways to add a route:

Adding a route directly with FastAPI's instance:

app = FastAPI()

@app.get("/{name}")
async def hi(name: str):
    return {"hi": name}

Or using APIRouter that is used typically in larger apps:

app = FastAPI()
router = APIRouter(prefix="/v1")

@router.get("/compliments/{name}")
async def hi1(name: str):
    return {"hi": name}

app.include_router(router)

Since we are trying to understand how FastAPI works, lets see what is happening when we use @app.{verb}:

    def get(
        self,
        path: Annotated[
            str,
            Doc("... # docs here"),
        ],
        *,
        ... # other args here
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
        return self.router.get(
            path,
            ... # code continues
        )

What we can see here is that FastAPI.{get,put,post,etc} are simply decorators that will include the path to APIRouter.

What about FastAPI.include_router?

FastAPI's function include_router will simply call it's own APIRouter's include_router, that basically will iterate through all routes included in your APIRouter and add the route.

    # FastAPI include_router

    def include_router(
        self,
        router: Annotated[routing.APIRouter, Doc("The `APIRouter` to include.")],
        *,
        ... # other args
    ) -> None:
        self.router.include_router(
            router,
            ... # other args
        )

    # APIRouter include_router

    def include_router(
        self,
        router: Annotated["APIRouter", Doc("The `APIRouter` to include.")],
        ... # other args
    ) -> None:
        for route in router.routes:
            if isinstance(route, APIRoute):
                ... # some logic here

                self.add_api_route(
                    prefix + route.path,
                    route.endpoint,
                    ... # other args
                )

Looking at APIRouter.include_router we can see that it handles other type of routes, like Starlette's routes, APIWebSocketRoute, etc...

And when my route function gets called?

When we receive a request, Starlette's Router will be called, since APIRouter don't override __call__. If it finds a matching route, it will call the route's handle function.

handle belongs to Route too, since it is not overwritten as well. What APIRoute does is setting Route's app to Starlette's function request_response, receiving APIRoute's get_route_handler as a parameter.

class APIRoute(routing.Route):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        ... # other args
    ) -> None:
        ... # some logic here
        self.app = request_response(self.get_route_handler())

get_route_handler returns the get_request_handler function. It's here that we start to see the "translation" of Starlette's request to a FastAPI route with dependants, pydantic models, etc.

It will run the run_endpoint_function function. And here is where our route function is being called with all the resolved dependencies, pydantic models, etc.

def get_request_handler(
    ... # args
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
    # logic here

    async def app(request: Request) -> Response:
        response: Union[Response, None] = None
        async with AsyncExitStack() as file_stack:
            # logic here
            errors: List[Any] = []
            async with AsyncExitStack() as async_exit_stack:
                # logic here
                if not errors:
                    raw_response = await run_endpoint_function(
                        dependant=dependant, values=values, is_coroutine=is_coroutine
                    )

Pretty cool the see how the framework you are using handles your code right?

In the next post, we'll take a look where and when FastAPI handles your API documentation.

Understanding FastAPI: How Starlette works

Rafael de Oliveira Marques — Mon, 24 Jun 2024 12:44:18 +0000

This post lives in:

We've seen in the last post that FastAPI is built on top of Starllete. We also saw that Starlette is a lightweight ASGI framework/toolkit.

Now, to start undertanding how FastAPI works, let's see what Starlette has to offer, how he deals with our HTTP requests, etc.

⚠️ Note: To work with starlette, you'll need at least python 3.8, so check which version you have installed on your computer before continuing

Hello world Starlette

Let's recreate that simple hello world from the previous post using Starllete:

from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


async def hello(request):
    return PlainTextResponse("Hello, World!")


app = Starlette(routes=[
    Route('/', hello),
])

Can you see the similarities with the example in the previous post? Let's remember it:

class SimplestFrameworkEver:
    async def __call__(self, scope, receive, send):
        await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [
            [b"content-type", b"text/plain"],
        ],

        })
        await send({
            "type": "http.response.body",
            "body": b"Hello, World!",
        })


app = SimplestFrameworkEver()

In both cases we have a class (SimplestFrameworkEver or Starlette), we create an instance of this class and pass it to the ASGI server to deal with it.

And according with ASGI's specification, an ASGI must expose a a single, asynchronous callable who receives a dictionary named scope and two other async callables named receive and send as parameters.

So if we are passing a Starlette object to the ASGI server, it MUST be a class that has a __call__ method implemented. Let's open Starlette's source code to check if it's true.

If we go to its official repository, we can find Starllete class inside starlette/applications.py:

async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
    scope["app"] = self
    if self.middleware_stack is None:
        self.middleware_stack = self.build_middleware_stack()
    await self.middleware_stack(scope, receive, send)

So we can see here that Starlette is a "callable" class, and apparently has a list of middlewares that will be executed each request. Simple as that, right? Yes and no 🤣

Things are never as simple as they seem, and we'll need to see what is a middleware_stack

ASGIApp

What we've seen this far is: Starlette is a callable that receives a scope, a receive and a send parameters. Which is exactly what and ASGI application is supposed to be. And what is a middleware_stack?

self.middleware_stack: ASGIApp | None = None

middleware_stack is an ASGIApp. But if you look in Starlette's types, ASGIApp is just an ASGI callable:

ASGIApp = typing.Callable[[Scope, Receive, Send], typing.Awaitable[None]]

If we take a look at Starlette.build_middleware_stack, we'll se a strange piece of code:

middleware = (
    [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
    + self.user_middleware
    + [
        Middleware(
            ExceptionMiddleware, handlers=exception_handlers, debug=debug
        )
    ]
)

app = self.router
for cls, args, kwargs in reversed(middleware):
    app = cls(app=app, *args, **kwargs)
return app

What is happening here is that starlette is creating sort of a chain of responsability of middlewares, and lastly our router / endpoint. Things will work like:

-> ServerErrorMiddleware
    -> Other Middlewares
        -> ExceptionMiddleware
            -> Router

Each ASGIApp will receive another ASGIApp as a dependency, and each one will call the next app when it gets called, till we reach ExceptionMiddleware, that will wrap our Router to deal with our exceptions.

The Router then will not receive any ASGIApp as a dependency. It will implement it's own app function (ASGI app) to match a route and execute our path operation function:

async def app(self, scope: Scope, receive: Receive, send: Send) -> None:
    # ... previous code

    for route in self.routes:
        # Determine if any route matches the incoming scope,
        # and hand over to the matching route if found.
        match, child_scope = route.matches(scope)
        if match == Match.FULL:
            scope.update(child_scope)
            await route.handle(scope, receive, send)
            return

    # ... code continues

Now that we know the flow of a Starlette's request, we can create a simple middleware to log the resquest's path, and another one that logs that everything went ok after the response is sent:

class LogRequestMiddleware:
    def __init__(self, app: ASGIApp):
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        logging.info(f"-> received a request @ {scope['path']}")
        await self.app(scope, receive, send)


class LogResponseMiddleware:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app

    async def __call__(self, scope: Scope, receive: Receive, send: Send):
        await self.app(scope, receive, send)
        logging.info("-> wow, we did it")


async def hello(request):
    logging.info("Great news, we got a request!")
    return PlainTextResponse("Hello, World!")


app = Starlette(
    routes=[
        Route('/', hello),
    ],
    middleware=[
        Middleware(LogRequestMiddleware),
        Middleware(LogResponseMiddleware)
    ]
)

And we'll get:

INFO:root:-> received a request @ /
INFO:root:Great news, we got a request!
INFO:     127.0.0.1:51770 - "GET / HTTP/1.1" 200 OK
INFO:root:-> wow, we did it

To learn more about Starlette's middlewares, you can read it's own documentation: Middleware. The people from the project made a great work documenting it.

Routes and Router

And last but not least, after all the chain of middlewares, we'll get our Router beeing executed wrapped in an ExceptionMiddleware, so it can deal with our exceptions. Router will have a list of routes to deal with.

If the router finds a matching route, it will call the route's handle function. The handle function will call our endpoint, that is basically the function or class that you passed while creating the Starlette app:

app = Starlette(
    routes=[
        Route('/', hello), # -> hello is the function that will be handled by Router's handle
    ],

Infact, a Router is an ASGIApp too, and you can dismiss all Starlette's middlewares by creating only a Router:

app = Router(routes=[
    Route('/', hello)
])

And what about FastAPI?

But Rafael, the title of your posts are saying Understanding FastAPI, but it's the second post and you keep writing stuff about ASGI specs, Starlette, etc.

But remember what I said: FastAPI is Starlette. It's a framework built on top of Starllete, and if you go to FastAPI's source code, you'll find this:

class FastAPI(Starlette):

    def __init__(
    # ... code continues

All this time you were reading about FastAPI, but I was writing about the internals of FastAPI, the foundations that support FastAPI.

In the next post we'll look how FastAPI extends Starlette and ASGI specs to offer us Pydantic models, OpenAPI specs out of the box, etc.

Stay tuned ;)

Understanding FastAPI: The Basics

Rafael de Oliveira Marques — Sat, 22 Jun 2024 22:47:41 +0000

This post lives in:

I've been working with FastAPI for a while now, and I decided to start digging a little deeper on it's internals.

Let's start from the beginning:

What is FastAPI?

As the docs says:

FastAPI is a modern, fast (high-performance), web framework for building APIs with Python based on standard Python type hints.

So here we have a good idea about what is FastAPI: A web framework for building APIs.

But it's not a framework built from scratch, it's a framework that was built on top of another framework: Starlette.

And what is Starlette?

If we go to starlette's website, we'll find that:

Starlette is a lightweight ASGI framework/toolkit, which is ideal for building async web services in Python.

And what is ASGI?

ASGI, or Asynchronous Server Gateway Interface is a specification that proposes an interface between web servers and python applications.

When we are running our fastapi application, we're using an ASGI server that will forward the request to our app.

Some of the most well-known asgi servers are:

Let's recap

FastAPI is a modern webframework written in python that is built on top Starlette, which in turn is a lightweight ASGI framework that needs an ASGI server to run.

Let's create a basic ASGI application

So if we want to understand how FastAPI works, we must start from the beginning: a simple asgi application:

async def app(scope, receive, send):
    await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [
            [b"content-type", b"text/plain"],
        ],
    })
    await send({
        "type": "http.response.body",
        "body": b"Hello, World!",
    })

Can you imagine that all features that FastAPI offers you, all middlewares and error handling, OpenAPI docs, etc., starts from this?

That's how ASGI is specified: a single asynchronous callable that takes a dict and two asynchronous callables as parameters.

And it works, without any webframework installed! Now let's check if it's really working:

First, install an ASGI server:

pip install uvicorn

And create a python file called app with the code above.

Let's run it and see if it's working:

uvicorn app:app

Now enter http://localhost:8000 in your browser:

INFO:     Started server process [4808]
INFO:     Waiting for application startup.
INFO:     ASGI 'lifespan' protocol appears unsupported.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:64045 - "GET / HTTP/1.1" 200 OK

Let's change to hypercorn to see if we are not tied to uvicorn somehow:

pip install hypercorn

And we can see it still works:

hypercorn app:app
[2024-06-22 19:00:55 -0300] [13008] [WARNING] ASGI Framework Lifespan error, continuing without Lifespan support
[2024-06-22 19:00:55 -0300] [13008] [INFO] Running on http://127.0.0.1:8000 (CTRL + C to quit)

Creating the simplest FastAPI clone ever

Now that we know what's beneath FastAPI, we can start playing a little bit and create the smallest, simpler ASGI framework ever:

class SimplestFrameworkEver:
    async def __call__(self, scope, receive, send):
        await send({
        "type": "http.response.start",
        "status": 200,
        "headers": [
            [b"content-type", b"text/plain"],
        ],

        })
        await send({
            "type": "http.response.body",
            "body": b"Hello, World!",
        })


app = SimplestFrameworkEver()

And you can still run it like before, with:

hypercorn app:app

uvicorn app:app

Next steps

Now that we can understand the basics on which FastAPI is made, we can move on to see:

how Starllete works
how FastAPI extends Starllete

Stay tuned for the next posts ;)

Python serialization with Marshmallow

Rafael de Oliveira Marques — Tue, 12 Jan 2021 13:41:17 +0000

When we are creating a API, we need a solid way to serialize and deserialize our data. The are lots of ways to do it, but in this article, we are going to see how to work with marshmallow.

So, let's take a look on how we start using marshmallow in our projects.

Install Marshmallow

To install marshmallow, we will use pip:

$ pip install -U marshmallow

Note that the -U parameter is used to upgrade all the packages to the newest version.

Now we have:

pip install -U marshmallow
Collecting marshmallow
Using cached marshmallow-3.5.1-py2.py3-none-any.whl (45 kB)
Installing collected packages: marshmallow
Successfully installed marshmallow-3.5.1

Creating our first Schema

To create a schema, we will use basically two imports: Schema and fields

To create a class representing our data, we will make it a subclass of Schema. To represent our fields, we will use the fields.TYPE, as you can see in the example above.

Now, we can serialize and deserialize our data using our schema:

As you can see, we basically use load to deserialize our data into python objects, and dump to serialize it again.

Validating data with Marshmallow

One of the great advantages of using a serialization library, is to be able to validate the incoming data. Marshmallow offers you a simple way of doing it.

As you can see in the code above, we sent an integer to a field expecting a valid UUID. When we use the load method, marshmallow will raise a ValidationError with all the data that are wrong.

You can use the messages property to check all data that failed to validate, and with valid_data property, you can check all data that is valid.

Custom message validations in marshmallow

Although we can use default validation messages provided by Marshmallow, we can provide our own custom messages:

And when we try to load the data, we’ll get our custom message to each cenario: