Making a Todo API with FastAPI and MongoDB

Introduction

This guide walks you through building the most up to date API with FastAPI application with MongoDB. You’ll create a Todo API that demonstrates best practices for structuring a FastAPI project, implementing CRUD operations, handling data validation, and organizing code for scalability.

By following this tutorial, you’ll learn how to:

• Set up a FastAPI project with proper tooling and dependencies
• Design and implement data models using Pydantic
• Create CRUD endpoints with async functions
• Connect to MongoDB using Motor
• Implement proper error handling and status codes
• Structure your project using routers and utilities
• Add logging and monitoring
• Configure CORS for frontend integration

Whether you’re new to FastAPI or looking to improve your existing skills, this guide provides a practical, hands-on approach to building APIs. The Todo application serves as a foundation that you can build upon for your own projects.

⭐️ The complete source code referenced in this guide is available on GitHub: https://github.com/zpillsbury/to-do
(There will be extra, code, I’ll be releasing how to add security and other features next)

Setup

• https://docs.astral.sh/uv/getting-started/installation/

• https://www.mongodb.com/try/download/community

Install uv

Why Use uv?

While not strictly required, the uv package manager offers significantly faster installation speeds compared to traditional tools like pip. It’s recommended to integrate uv into your projects moving forward.

Create basic project setup

$ uv add --dev black mypy ruff
$ uv add --dev black mypy ruff
$ uv add --dev black mypy ruff

Enter fullscreen mode Exit fullscreen mode

This command installs:

black: Code formatter
mypy: Static type checker
ruff: Linter
These tools help maintain code quality, consistency, and readability.

$ uv add --dev black mypy ruff
$ uv add --dev black mypy ruff
$ uv add --dev black mypy ruff

Enter fullscreen mode Exit fullscreen mode

Add tool settings in pyproject.toml for black, ruff, and mypy:

pyproject.toml

<span>tool</span><span>.</span><span>black</span><span>]</span>
<span>line</span><span>-</span><span>length</span> <span>=</span> <span>88</span>
<span>[</span><span>tool</span><span>.</span><span>ruff</span><span>]</span>
<span>lint</span><span>.</span><span>select</span> <span>=</span> <span>[</span><span>"</span><span>E</span><span>"</span><span>,</span> <span>"</span><span>F</span><span>"</span><span>,</span> <span>"</span><span>I</span><span>"</span><span>]</span>
<span>lint</span><span>.</span><span>fixable</span> <span>=</span> <span>[</span><span>"</span><span>ALL</span><span>"</span><span>]</span>
<span>exclude</span> <span>=</span> <span>[</span><span>"</span><span>.git</span><span>"</span><span>,</span> <span>"</span><span>.mypy_cache</span><span>"</span><span>,</span> <span>"</span><span>.ruff_cache</span><span>"</span><span>]</span>
<span>line</span><span>-</span><span>length</span> <span>=</span> <span>88</span>
<span>[</span><span>tool</span><span>.</span><span>mypy</span><span>]</span>
<span>plugins</span> <span>=</span> <span>[</span><span>"</span><span>pydantic.mypy</span><span>"</span><span>]</span>
<span>disallow_any_generics</span> <span>=</span> <span>true</span>
<span>disallow_subclassing_any</span> <span>=</span> <span>true</span>
<span>disallow_untyped_calls</span> <span>=</span> <span>true</span>
<span>disallow_untyped_defs</span> <span>=</span> <span>true</span>
<span>disallow_incomplete_defs</span> <span>=</span> <span>true</span>
<span>check_untyped_defs</span> <span>=</span> <span>true</span>
<span>no_implicit_optional</span> <span>=</span> <span>true</span>
<span>warn_redundant_casts</span> <span>=</span> <span>true</span>
<span>warn_unused_ignores</span> <span>=</span> <span>true</span>
<span>warn_return_any</span> <span>=</span> <span>true</span>
<span>strict_equality</span> <span>=</span> <span>true</span>
<span>disallow_untyped_decorators</span> <span>=</span> <span>false</span>
<span>ignore_missing_imports</span> <span>=</span> <span>true</span>
<span>implicit_reexport</span> <span>=</span> <span>true</span>
<span>[</span><span>tool</span><span>.</span><span>pydantic</span><span>-</span><span>mypy</span><span>]</span>
<span>init_forbid_extra</span> <span>=</span> <span>true</span>
<span>init_typed</span> <span>=</span> <span>true</span>
<span>warn_required_dynamic_aliases</span> <span>=</span> <span>true</span>
<span>warn_untyped_fields</span> <span>=</span> <span>true</span>
<span>tool</span><span>.</span><span>black</span><span>]</span>
<span>line</span><span>-</span><span>length</span> <span>=</span> <span>88</span>

<span>[</span><span>tool</span><span>.</span><span>ruff</span><span>]</span>
<span>lint</span><span>.</span><span>select</span> <span>=</span> <span>[</span><span>"</span><span>E</span><span>"</span><span>,</span> <span>"</span><span>F</span><span>"</span><span>,</span> <span>"</span><span>I</span><span>"</span><span>]</span>
<span>lint</span><span>.</span><span>fixable</span> <span>=</span> <span>[</span><span>"</span><span>ALL</span><span>"</span><span>]</span>
<span>exclude</span> <span>=</span> <span>[</span><span>"</span><span>.git</span><span>"</span><span>,</span> <span>"</span><span>.mypy_cache</span><span>"</span><span>,</span> <span>"</span><span>.ruff_cache</span><span>"</span><span>]</span>
<span>line</span><span>-</span><span>length</span> <span>=</span> <span>88</span>

<span>[</span><span>tool</span><span>.</span><span>mypy</span><span>]</span>
<span>plugins</span> <span>=</span> <span>[</span><span>"</span><span>pydantic.mypy</span><span>"</span><span>]</span>

<span>disallow_any_generics</span> <span>=</span> <span>true</span>
<span>disallow_subclassing_any</span> <span>=</span> <span>true</span>
<span>disallow_untyped_calls</span> <span>=</span> <span>true</span>
<span>disallow_untyped_defs</span> <span>=</span> <span>true</span>
<span>disallow_incomplete_defs</span> <span>=</span> <span>true</span>
<span>check_untyped_defs</span> <span>=</span> <span>true</span>
<span>no_implicit_optional</span> <span>=</span> <span>true</span>
<span>warn_redundant_casts</span> <span>=</span> <span>true</span>
<span>warn_unused_ignores</span> <span>=</span> <span>true</span>
<span>warn_return_any</span> <span>=</span> <span>true</span>
<span>strict_equality</span> <span>=</span> <span>true</span>
<span>disallow_untyped_decorators</span> <span>=</span> <span>false</span>
<span>ignore_missing_imports</span> <span>=</span> <span>true</span>
<span>implicit_reexport</span> <span>=</span> <span>true</span>

<span>[</span><span>tool</span><span>.</span><span>pydantic</span><span>-</span><span>mypy</span><span>]</span>
<span>init_forbid_extra</span> <span>=</span> <span>true</span>
<span>init_typed</span> <span>=</span> <span>true</span>
<span>warn_required_dynamic_aliases</span> <span>=</span> <span>true</span>
<span>warn_untyped_fields</span> <span>=</span> <span>true</span>
tool.black]
line-length = 88

[tool.ruff]
lint.select = ["E", "F", "I"]
lint.fixable = ["ALL"]
exclude = [".git", ".mypy_cache", ".ruff_cache"]
line-length = 88

[tool.mypy]
plugins = ["pydantic.mypy"]

disallow_any_generics = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_return_any = true
strict_equality = true
disallow_untyped_decorators = false
ignore_missing_imports = true
implicit_reexport = true

[tool.pydantic-mypy]
init_forbid_extra = true
init_typed = true
warn_required_dynamic_aliases = true
warn_untyped_fields = true

Enter fullscreen mode Exit fullscreen mode

Synchronize the new configurations:

$ uv sync
$ uv sync
$ uv sync

Enter fullscreen mode Exit fullscreen mode

Installing FastAPI:

$ uv add "fastapi[standard]"
$ uv add "fastapi[standard]"
$ uv add "fastapi[standard]"

Enter fullscreen mode Exit fullscreen mode

Create an .env file to store your secrets, make sure to put it in your .gitignore. This makes it where any file put inside there won’t be pushed up to github.
You will need to get your mongo uri from your MongoDB.

.env

<span>MONGO_URI</span><span>=</span><span>mongodb</span><span>:</span><span>//</span><span>root</span><span>:</span><span>mySecureDbPassword1</span><span>@localhost</span><span>:</span><span>27017</span><span>/</span>
<span>MONGO_URI</span><span>=</span><span>mongodb</span><span>:</span><span>//</span><span>root</span><span>:</span><span>mySecureDbPassword1</span><span>@localhost</span><span>:</span><span>27017</span><span>/</span>
MONGO_URI=mongodb://root:mySecureDbPassword1@localhost:27017/

Enter fullscreen mode Exit fullscreen mode

Delete the hello.py and create main.py.

Use the Pydantic BaseSettings to load the mongo_uri from your .env folder, we will be using SecretSTR so the information isn’t visible in logs or tracebacks.

main.py

<span>from</span> <span>pydantic</span> <span>import</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>
<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretSTR</span>
    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>
<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>

<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretSTR</span>

    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>


<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretSTR

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

Enter fullscreen mode Exit fullscreen mode

Next we will be making a function get_db to connect to MongoDB using Motor which is an asynchronous drive. In the function we will be calling settings to get the mongo uri.

The ["to-do"] will be the name of your database, you don’t need to create it in mongo, once you create your first todo in your FastAPI it will generate the database and collections you use.

We will be using asynchronous functions to make this API as fast as possible. What makes async so much faster is that functions will be able to be process other tasks while waiting for tasks like a database query or a call to an external api.

If you need a more in depth explanation of the benefits of async over sync click here: https://www.youtube.com/watch?v=8aGhZQkoFbQ

Add app = FASTAPI to be able to call FastAPI.

main.py

<span>from</span> <span>typing</span> <span>import</span> <span>Any</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span>
<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>
<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>
<span>app</span> <span>=</span> <span>FastAPI</span><span>()</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Any</span>

<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span>
<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>

<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>


<span>app</span> <span>=</span> <span>FastAPI</span><span>()</span>
from typing import Any

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

def get_db() -> AsyncIOMotorDatabase[Any]:
    """ Get MongoDB """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]


app = FastAPI()

Enter fullscreen mode Exit fullscreen mode

You should be able to call fastAPI now using the command FastAPI dev main.py.
For now it should be an empty page since we haven’t added any endpoints.

BaseModels

The first thing you need to do to make a good API is to set up your pydantic base models. At this phase you will set up the models that you will be building your API with. This is important to do before you start making your API, since these will be your guidelines when setting up your Create, Read, Update and Delete for your todos.

Read Model

Starting with Todo , this will include all information you want to retrieve from MongoDb. We will use this for our Get functions.

Since we are making a todo app with mongoDB, it will need the id . This is how we can search for a specific record from the mongo database.

We need a title this is what the task is going to be named.

We can add a description for tasks that might needs an explanation.

Finally we need completed this is where we will check off what we have done.

main.py

<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>
<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>

<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>
from pydantic import BaseModel

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

Enter fullscreen mode Exit fullscreen mode

Now we have the base model to build the rest of our models off of.

Create Model

The next base model we will make is CreateTodo. This will be all the information the user will enter theirselves.

Since Mongo supplies the id it won’t be needed here.

We will add in the title as a mandatory field, without this the todo can’t be made.

Next we will add description in but we don’t always want to add a description to everything. If you add “take out trash”, you don’t need a description.

This is where Optional comes in. This does exactly what it sounds like making it optional for you to either add a field or not.

Next the completed will be inserted in as automatically False by our function. So the user won’t have to do it themselves every time. This will be shown how to this later on.

Id Model

Next we will need to know the id of the todo we just created. So we will make a TodoId base model .

We will be using this to return the id whenever the user creates a new todo. This is helpful so the user doesn’t have to do a find all or go to the database to find the id of the new entry.

main.py

<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>
<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>

<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>


<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
from typing import Optional

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]


class TodoId(BaseModel):
    id: str

Enter fullscreen mode Exit fullscreen mode

Update Model

The next base model is UpdateTodo. This will be the fields we want our users to be able to update.

If they typed in the wrong title or description they can go update it to change it. More importantly to be able to change completed that way the Todo actually works.

The main problem is if you want to change one of these you don’t want to have to change all of them. This is where Optional comes in handy again, allowing fields in base models to either be changed or not.

Result Model

Now that we have the update model, when we actually update we will want to know if it was a successful update or not. This is where we will create SuccessResult , we will use this to return if the action was successful or not.

Since the Delete end point just needs to know if its succeeded or failed we will be using the same SuccessResult base model for the return.

main.py

<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>
<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>

<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class SuccessResult(BaseModel):
    success: bool

Enter fullscreen mode Exit fullscreen mode

All Models

main.py

<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>
<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>
<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>
<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>

<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>

<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>

<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>

<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>

<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
from pydantic import BaseModel
from typing import Optional

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

Enter fullscreen mode Exit fullscreen mode

CRUD

now that we have the project set up we can move to make the industry standard Create, Read, Update, and Delete(CRUD) functions based off the models.

Create

To create a new Todo, start by defining an endpoint with @app.post("/todo"). This specifies the route for FastAPI and marks it as a POST endpoint, which is used for creating resources.

Next, create an asynchronous function create_to_do that takes new_todo: CreateTodo as a parameter. This parameter is a Pydantic model representing the input data for creating a Todo. The function will return a TodoId model containing the ID of the newly created Todo.

To prepare the data for insertion, use the .model_dump() method to convert the new_todo model into a dictionary. MongoDB requires documents in dictionary format, and you can add default fields, like {“completed”: False}, using the dictionary merge operator (|).

Insert the data into the todos collection using the db.todos.insert_one(data) method. If the collection doesn’t already exist, MongoDB will create it automatically. The await keyword ensures that the function can continue handling other tasks while waiting for the database operation to complete.

Finally, retrieve the inserted_id from MongoDB, convert it to a string (as JSON only supports strings and numbers), and return it.

you will notice I broke down each part into separate variables, particularly when you are working with other people its important to make it as readable as possible. This will also help you in the future to be able to find bugs easier.

main.py

<span>@app.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>
    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>
    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>
    <span>return</span> <span>TodoID</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>
<span>@app.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>

    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>

    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>

    <span>return</span> <span>TodoID</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>
@app.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """ Create a new todo """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoID(id=id)

Enter fullscreen mode Exit fullscreen mode

go to fastapi dev main.py to try it out, you should now be able to create your first todo. Once you create it, you can check in MongoDB if it was successful. The next step is going to be making it where we can get that information without having to check Mongo ourselves.

Read

Get All Todos

The get_all_todos endpoint retrieves all todo items from the database. you will notice you will be returning a list of Todos.

you will make a variable with an empty list, such as result . We will be using it for the function below, we will be looping for each data entry(todo) in todos then appending it to result .

Once the loop is done you will return result which now is a list of Todos.

main.py

<span>@app.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>_id</span><span>"</span><span>]),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>title</span><span>"</span><span>],</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>description</span><span>"</span><span>],</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>completed</span><span>"</span><span>],</span>
            <span>)</span>
        <span>)</span>
    <span>return</span> <span>result</span>
<span>@app.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>_id</span><span>"</span><span>]),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>title</span><span>"</span><span>],</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>description</span><span>"</span><span>],</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>completed</span><span>"</span><span>],</span>
            <span>)</span>
        <span>)</span>

    <span>return</span> <span>result</span>
@app.get("/todo")
async def get_all_todos() -> list[Todo]:
    """ Get all todos """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get["_id"]),
                title=todo.get["title"],
                description=todo.get["description"],
                completed=todo.get["completed"],
            )
        )

    return result

Enter fullscreen mode Exit fullscreen mode

Try it out, you should be able to see your todo you made.try make another todo you will be able to get both of them. If you want to only get one you can get todo by id. Which will be covered next.

Get Todo by ID

Since I’ve explained the process in depth on the last function, I will only hit the main points and new things moving forward.

The get_one_todo endpoint gets a single todo item by its ID.

The provided todo_id string is converted into a MongoDB ObjectId.

We then use find_one method searches for the document with the matching _id.

It returns the Todo if the id matches.

If no document is found, an HTTP 404 error is raised.

Main.py

<span>@app.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>
    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>_id</span><span>"</span><span>]),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>title</span><span>"</span><span>],</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>description</span><span>"</span><span>],</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>completed</span><span>"</span><span>],</span>
    <span>)</span>
<span>@app.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>

    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>_id</span><span>"</span><span>]),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>title</span><span>"</span><span>],</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>description</span><span>"</span><span>],</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>[</span><span>"</span><span>completed</span><span>"</span><span>],</span>
    <span>)</span>
@app.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """ Get todo by ID """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get["_id"]),
        title=todo.get["title"],
        description=todo.get["description"],
        completed=todo.get["completed"],
    )

Enter fullscreen mode Exit fullscreen mode

Try it out, grab one of your ids and see if you can find your todo. If it works great! you will see we can do multiple things with an id .

Update

Update Todo

The update_one_todo endpoint updates the fields of an existing todo item.

The UpdateTodo model allows users to provide only the fields they want to change.

You will use update_one to apply changes.

If successful it shows success=True

If no matching document is found, an HTTP 404 error is raised.

main.py

<span>@app.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>
    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>
    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>@app.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>

    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>

    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
@app.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """ Update todo by ID """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Enter fullscreen mode Exit fullscreen mode

Try it out, if you see a success and you want to verify it worked. Take the id and put it in the get_one_todo. You should see that the information is updated on your todo.

Delete

Delete Todo

The delete_one_todo endpoint removes a todo item from the database.

MongoDB’s delete_one method removes the matching document.

If successful it shows success=True

If no document is deleted, an HTTP 404 error is raised.

main.py

<span>@app.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>@app.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>

    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
@app.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """ Delete todo by ID """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Enter fullscreen mode Exit fullscreen mode

Try it out, if its a success you can take the id to the get_one_todo to verify. It should say Todo not found.

Utilities

Logging

Logging allows for troubleshooting by monitoring your application in real time. This is important to be able to fix and bugs when they occur. We will just use the basic python logging module.

logging configs:

main.py

<span>import</span> <span>logging</span>
<span>import</span> <span>sys</span>
<span>logging</span><span>.</span><span>basicConfig</span><span>(</span>
    <span>stream</span><span>=</span><span>sys</span><span>.</span><span>stdout</span><span>,</span>
    <span>level</span><span>=</span><span>logging</span><span>.</span><span>INFO</span><span>,</span>
    <span>format</span><span>=</span><span>"</span><span>[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s</span><span>"</span><span>,</span>  <span># noqa: E501 </span>    <span>datefmt</span><span>=</span><span>"</span><span>%d/%b/%Y %H:%M:%S</span><span>"</span><span>,</span>
<span>)</span>
<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>"</span><span>todo</span><span>"</span><span>)</span>
<span>import</span> <span>logging</span>
<span>import</span> <span>sys</span>

<span>logging</span><span>.</span><span>basicConfig</span><span>(</span>
    <span>stream</span><span>=</span><span>sys</span><span>.</span><span>stdout</span><span>,</span>
    <span>level</span><span>=</span><span>logging</span><span>.</span><span>INFO</span><span>,</span>
    <span>format</span><span>=</span><span>"</span><span>[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s</span><span>"</span><span>,</span>  <span># noqa: E501 </span>    <span>datefmt</span><span>=</span><span>"</span><span>%d/%b/%Y %H:%M:%S</span><span>"</span><span>,</span>
<span>)</span>
<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>"</span><span>todo</span><span>"</span><span>)</span>
import logging
import sys

logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s",  # noqa: E501     datefmt="%d/%b/%Y %H:%M:%S",
)
logger = logging.getLogger("todo")

Enter fullscreen mode Exit fullscreen mode

code breakdown:

stream=sys.stdout :

This makes it where it outputs to you console

level=logging.INFO :

This sets the minimum log level

it goes Debug → Info → Warning → Error → Critical

format="..." :

[%(asctime)s] : The timestamp of the log

%(levelname)s : The severity level

[%(name)s.%(funcName)s:%(lineno)d]

%(name)s : Name of the Logger

%(funcName)s : Name of the function where the logger was created

%(lineno)d : The line where the log was created

%(message)s : The log message

datefmt="%d/%b/%Y %H:%M:%S”

The format the timestamp will be in.

Middleware

Middleware in FastAPI allows you to process requests and responses globally.
CORS Middleware

main.py

<span>from</span> <span>fastapi.middleware.cors</span> <span>import</span> <span>CORSMiddleware</span>
<span>app</span><span>.</span><span>add_middleware</span><span>(</span>
    <span>CORSMiddleware</span><span>,</span>
    <span>allow_credentials</span><span>=</span><span>True</span><span>,</span>
    <span>allow_methods</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_headers</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_origins</span><span>=</span><span>[</span>
        <span>"</span><span>http://localhost:3000</span><span>"</span><span>,</span>
        <span>"</span><span>http://localhost:8000</span><span>"</span><span>,</span>
    <span>],</span>
<span>)</span>
<span>from</span> <span>fastapi.middleware.cors</span> <span>import</span> <span>CORSMiddleware</span>
<span>app</span><span>.</span><span>add_middleware</span><span>(</span>
    <span>CORSMiddleware</span><span>,</span>
    <span>allow_credentials</span><span>=</span><span>True</span><span>,</span>
    <span>allow_methods</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_headers</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_origins</span><span>=</span><span>[</span>
        <span>"</span><span>http://localhost:3000</span><span>"</span><span>,</span>
        <span>"</span><span>http://localhost:8000</span><span>"</span><span>,</span>
    <span>],</span>
<span>)</span>
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_origins=[
        "http://localhost:3000",
        "http://localhost:8000",
    ],
)

Enter fullscreen mode Exit fullscreen mode

Here’s a concise breakdown of the CORS middleware setup:

allow_credentials=True
Allows cookies, auth headers, and other credentials to be sent with requests .
allow_methods=["*"]
Permits all HTTP methods (e.g., GET, POST, DELETE) to be used by the frontend .
allow_headers=["*"]
Accepts all headers, providing flexibility for metadata or labels in requests and responses .
allow_origins=
Grants access to specific addresses, such as the backend http://localhost:8000 and the frontend http://localhost:3000 for security and proper integration.

main.py

<span>@app.middleware</span><span>(</span><span>"</span><span>http</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>process_time_log_middleware</span><span>(</span><span>request</span><span>:</span> <span>Request</span><span>,</span> <span>call_next</span><span>:</span> <span>F</span><span>)</span> <span>-></span> <span>Response</span><span>:</span>
    <span>"""</span><span> Add API process time in response headers and log calls </span><span>"""</span>
    <span>start_time</span> <span>=</span> <span>time</span><span>.</span><span>time</span><span>()</span>
    <span>response</span><span>:</span> <span>Response</span> <span>=</span> <span>await</span> <span>call_next</span><span>(</span><span>request</span><span>)</span>
    <span>process_time</span> <span>=</span> <span>str</span><span>(</span><span>round</span><span>(</span><span>time</span><span>.</span><span>time</span><span>()</span> <span>-</span> <span>start_time</span><span>,</span> <span>3</span><span>))</span>
    <span>response</span><span>.</span><span>headers</span><span>[</span><span>"</span><span>X-Process-Time</span><span>"</span><span>]</span> <span>=</span> <span>process_time</span>
    <span>logger</span><span>.</span><span>info</span><span>(</span>
        <span>"</span><span>Method=%s Path=%s StatusCode=%s ProcessTime=%s</span><span>"</span><span>,</span>
        <span>request</span><span>.</span><span>method</span><span>,</span>
        <span>request</span><span>.</span><span>url</span><span>.</span><span>path</span><span>,</span>
        <span>response</span><span>.</span><span>status_code</span><span>,</span>
        <span>process_time</span><span>,</span>
    <span>)</span>
    <span>return</span> <span>response</span>
<span>@app.middleware</span><span>(</span><span>"</span><span>http</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>process_time_log_middleware</span><span>(</span><span>request</span><span>:</span> <span>Request</span><span>,</span> <span>call_next</span><span>:</span> <span>F</span><span>)</span> <span>-></span> <span>Response</span><span>:</span>
    <span>"""</span><span> Add API process time in response headers and log calls </span><span>"""</span>
    <span>start_time</span> <span>=</span> <span>time</span><span>.</span><span>time</span><span>()</span>
    <span>response</span><span>:</span> <span>Response</span> <span>=</span> <span>await</span> <span>call_next</span><span>(</span><span>request</span><span>)</span>
    <span>process_time</span> <span>=</span> <span>str</span><span>(</span><span>round</span><span>(</span><span>time</span><span>.</span><span>time</span><span>()</span> <span>-</span> <span>start_time</span><span>,</span> <span>3</span><span>))</span>
    <span>response</span><span>.</span><span>headers</span><span>[</span><span>"</span><span>X-Process-Time</span><span>"</span><span>]</span> <span>=</span> <span>process_time</span>

    <span>logger</span><span>.</span><span>info</span><span>(</span>
        <span>"</span><span>Method=%s Path=%s StatusCode=%s ProcessTime=%s</span><span>"</span><span>,</span>
        <span>request</span><span>.</span><span>method</span><span>,</span>
        <span>request</span><span>.</span><span>url</span><span>.</span><span>path</span><span>,</span>
        <span>response</span><span>.</span><span>status_code</span><span>,</span>
        <span>process_time</span><span>,</span>
    <span>)</span>

    <span>return</span> <span>response</span>
@app.middleware("http")
async def process_time_log_middleware(request: Request, call_next: F) -> Response:
    """ Add API process time in response headers and log calls """
    start_time = time.time()
    response: Response = await call_next(request)
    process_time = str(round(time.time() - start_time, 3))
    response.headers["X-Process-Time"] = process_time

    logger.info(
        "Method=%s Path=%s StatusCode=%s ProcessTime=%s",
        request.method,
        request.url.path,
        response.status_code,
        process_time,
    )

    return response

Enter fullscreen mode Exit fullscreen mode

This middleware logs request details and tracks processing time for each API call

Captures the current time when a request is received to measure processing duration.
Passes the request to the next handler using call_next and waits for the response.
Determines the time taken for the request by subtracting the start time from the end time and rounding it to three decimal places.
Attaches the processing time as X-Process-Time in the response headers.
Logs the HTTP method, request path, status code, and processing time for better monitoring and debugging.

Organization

At this point your document should be looking like this:

Main.py

<span>from</span> <span>typing</span> <span>import</span> <span>Any</span><span>,</span> <span>Optional</span>
<span>from</span> <span>bson</span> <span>import</span> <span>ObjectId</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span><span>,</span> <span>HTTPException</span><span>,</span> <span>status</span>
<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span><span>,</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>
<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretStr</span>
    <span>token_url</span><span>:</span> <span>str</span>
    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>
<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>
<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>
<span>db</span> <span>=</span> <span>get_db</span><span>()</span>
<span>app</span> <span>=</span> <span>FastAPI</span><span>(</span><span>docs_url</span><span>=</span><span>"</span><span>/</span><span>"</span><span>)</span>
<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>
<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>
<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
<span>@app.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>
    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>
    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>
    <span>return</span> <span>TodoId</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>
<span>@app.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
            <span>)</span>
        <span>)</span>
    <span>return</span> <span>result</span>
<span>@app.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>
    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
    <span>)</span>
<span>@app.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>
    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>
    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>@app.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Any</span><span>,</span> <span>Optional</span>

<span>from</span> <span>bson</span> <span>import</span> <span>ObjectId</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span><span>,</span> <span>HTTPException</span><span>,</span> <span>status</span>
<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span><span>,</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>

<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretStr</span>
    <span>token_url</span><span>:</span> <span>str</span>

    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>

<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>

<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>

<span>db</span> <span>=</span> <span>get_db</span><span>()</span>

<span>app</span> <span>=</span> <span>FastAPI</span><span>(</span><span>docs_url</span><span>=</span><span>"</span><span>/</span><span>"</span><span>)</span>

<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>

<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>

<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>

<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>

<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>

<span>@app.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>

    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>

    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>

    <span>return</span> <span>TodoId</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>

<span>@app.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
            <span>)</span>
        <span>)</span>

    <span>return</span> <span>result</span>

<span>@app.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>

    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
    <span>)</span>

<span>@app.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>

    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>

    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>

<span>@app.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>

    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
from typing import Any, Optional

from bson import ObjectId
from fastapi import FastAPI, HTTPException, status
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretStr
    token_url: str

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

settings = Settings()

def get_db() -> AsyncIOMotorDatabase[Any]:
    """ Get MongoDB """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]

db = get_db()

app = FastAPI(docs_url="/")

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: str

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

@app.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """ Create a new todo """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoId(id=id)

@app.get("/todo")
async def get_all_todos() -> list[Todo]:
    """ Get all todos """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get("_id")),
                title=todo.get("title"),
                description=todo.get("description"),
                completed=todo.get("completed"),
            )
        )

    return result

@app.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """ Get todo by ID """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get("_id")),
        title=todo.get("title"),
        description=todo.get("description"),
        completed=todo.get("completed"),
    )

@app.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """ Update todo by ID """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

@app.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """ Delete todo by ID """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Enter fullscreen mode Exit fullscreen mode

As you can see its getting a bit long and if you wanted to add more end points, security, dockers and so on its only going to get longer and longer. This is where organization is important. The main tools we will be using for this is routers and folders.

Folder Setup

First we will make an app folder to hold everything.

Next we will move our main.py into the app folder.

We will be adding __init__.py to any folders that have files.

Next we will make 2 more folders routers and utils both are inside our app folder.

Utils

Next we will be adding config.py , settings.py, log.py and __init__.py to the utils folder.

Settings.py

You will go into main.py and be moving the basesettings to settings.py

You can copy and paste all the imports as well and use shift + command + p and use ruff to get rid of all unused imports on all files moving forward.

settings.py

<span>from</span> <span>pydantic</span> <span>import</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>
<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretStr</span>
    <span>token_url</span><span>:</span> <span>str</span>
    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>
<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>SecretStr</span>
<span>from</span> <span>pydantic_settings</span> <span>import</span> <span>BaseSettings</span><span>,</span> <span>SettingsConfigDict</span>

<span>class</span> <span>Settings</span><span>(</span><span>BaseSettings</span><span>):</span>
    <span>mongo_uri</span><span>:</span> <span>SecretStr</span>
    <span>token_url</span><span>:</span> <span>str</span>

    <span>model_config</span> <span>=</span> <span>SettingsConfigDict</span><span>(</span><span>env_file</span><span>=</span><span>"</span><span>.env</span><span>"</span><span>,</span> <span>extra</span><span>=</span><span>"</span><span>ignore</span><span>"</span><span>)</span>

<span>settings</span> <span>=</span> <span>Settings</span><span>()</span>
from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretStr
    token_url: str

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

settings = Settings()

Config.py

Make sure to import settings from .settings

config.py

<span>from</span> <span>typing</span> <span>import</span> <span>Any</span>
<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>
<span>from</span> <span>.settings</span> <span>import</span> <span>settings</span>
<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>
<span>db</span> <span>=</span> <span>get_db</span><span>()</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Any</span>

<span>from</span> <span>motor.motor_asyncio</span> <span>import</span> <span>AsyncIOMotorClient</span><span>,</span> <span>AsyncIOMotorDatabase</span>

<span>from</span> <span>.settings</span> <span>import</span> <span>settings</span>

<span>def</span> <span>get_db</span><span>()</span> <span>-></span> <span>AsyncIOMotorDatabase</span><span>[</span><span>Any</span><span>]:</span>
    <span>"""</span><span> Get MongoDB </span><span>"""</span>
    <span>return</span> <span>AsyncIOMotorClient</span><span>(</span><span>settings</span><span>.</span><span>mongo_uri</span><span>.</span><span>get_secret_value</span><span>())[</span><span>"</span><span>to-do</span><span>"</span><span>]</span>

<span>db</span> <span>=</span> <span>get_db</span><span>()</span>
from typing import Any

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

from .settings import settings

def get_db() -> AsyncIOMotorDatabase[Any]:
    """ Get MongoDB """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]

db = get_db()

Enter fullscreen mode Exit fullscreen mode

log.py

log.py

<span>import</span> <span>logging</span>
<span>import</span> <span>sys</span>
<span>logging</span><span>.</span><span>basicConfig</span><span>(</span>
    <span>stream</span><span>=</span><span>sys</span><span>.</span><span>stdout</span><span>,</span>
    <span>level</span><span>=</span><span>logging</span><span>.</span><span>INFO</span><span>,</span>
    <span>format</span><span>=</span><span>"</span><span>[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s</span><span>"</span><span>,</span>  <span># noqa: E501 </span>    <span>datefmt</span><span>=</span><span>"</span><span>%d/%b/%Y %H:%M:%S</span><span>"</span><span>,</span>
<span>)</span>
<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>"</span><span>todo</span><span>"</span><span>)</span>
<span>import</span> <span>logging</span>
<span>import</span> <span>sys</span>

<span>logging</span><span>.</span><span>basicConfig</span><span>(</span>
    <span>stream</span><span>=</span><span>sys</span><span>.</span><span>stdout</span><span>,</span>
    <span>level</span><span>=</span><span>logging</span><span>.</span><span>INFO</span><span>,</span>
    <span>format</span><span>=</span><span>"</span><span>[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s</span><span>"</span><span>,</span>  <span># noqa: E501 </span>    <span>datefmt</span><span>=</span><span>"</span><span>%d/%b/%Y %H:%M:%S</span><span>"</span><span>,</span>
<span>)</span>
<span>logger</span> <span>=</span> <span>logging</span><span>.</span><span>getLogger</span><span>(</span><span>"</span><span>todo</span><span>"</span><span>)</span>
import logging
import sys

logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s",  # noqa: E501     datefmt="%d/%b/%Y %H:%M:%S",
)
logger = logging.getLogger("todo")

Enter fullscreen mode Exit fullscreen mode

Routers

Next we will be adding todos folder in routers

Inside todos folder we will add todo.py , models.py , and __**init**__.py .

Models.py

We now now go into main.py and cut out all the models and and paste them into models.py .

models.py should now look like:

models.py

<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>
<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>
<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>
<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
<span>from</span> <span>pydantic</span> <span>import</span> <span>BaseModel</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Optional</span>

<span>class</span> <span>Todo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>str</span>
    <span>completed</span><span>:</span> <span>bool</span>

<span>class</span> <span>CreateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>str</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>

<span>class</span> <span>UpdateTodo</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>title</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>description</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span>
    <span>completed</span><span>:</span> <span>Optional</span><span>[</span><span>bool</span><span>]</span>

<span>class</span> <span>TodoId</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>id</span><span>:</span> <span>str</span>

<span>class</span> <span>SuccessResult</span><span>(</span><span>BaseModel</span><span>):</span>
    <span>success</span><span>:</span> <span>bool</span>
from pydantic import BaseModel
from typing import Optional

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

Enter fullscreen mode Exit fullscreen mode

Todo.py

We will now be moving all the CRUD functions over too todo.py .

You will notice some problems with you code now. Your models you use in the function are no longer there and @app no longer works.

To fix that add from .models import CreateTodo, SuccessResult, Todo, TodoId, UpdateTodo

Add from app.utils.conif import db

We will then add in our router , to do that we just add in router = APIRouter() . We can use a function of prefix= to make it where we don’t have to type the path for each function in the router, you will still have to add {todo_id} for functions that need it.

As well as add in tags= which sorts out your APIs into categories. This will be more important when you have more routers.

We then replace @app with @router .

todo.py should look like this:

todo.py

<span>from</span> <span>bson</span> <span>import</span> <span>ObjectId</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>APIRouter</span><span>,</span> <span>HTTPException</span><span>,</span> <span>status</span>
<span>from</span> <span>app.utils.config</span> <span>import</span> <span>db</span>
<span>from</span> <span>.models</span> <span>import</span> <span>CreateTodo</span><span>,</span> <span>SuccessResult</span><span>,</span> <span>Todo</span><span>,</span> <span>TodoId</span><span>,</span> <span>UpdateTodo</span>
<span>@router.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>
    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>
    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>
    <span>return</span> <span>TodoId</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>
<span>@router.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
            <span>)</span>
        <span>)</span>
    <span>return</span> <span>result</span>
<span>@router.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>
    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
    <span>)</span>
<span>@router.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>
    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>
    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>@router.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>
    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>
    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
<span>from</span> <span>bson</span> <span>import</span> <span>ObjectId</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>APIRouter</span><span>,</span> <span>HTTPException</span><span>,</span> <span>status</span>

<span>from</span> <span>app.utils.config</span> <span>import</span> <span>db</span>

<span>from</span> <span>.models</span> <span>import</span> <span>CreateTodo</span><span>,</span> <span>SuccessResult</span><span>,</span> <span>Todo</span><span>,</span> <span>TodoId</span><span>,</span> <span>UpdateTodo</span>

<span>@router.post</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>create_to_do</span><span>(</span><span>new_todo</span><span>:</span> <span>CreateTodo</span><span>)</span> <span>-></span> <span>TodoId</span><span>:</span>
    <span>"""</span><span> Create a new todo </span><span>"""</span>
    <span>data</span> <span>=</span> <span>new_todo</span><span>.</span><span>model_dump</span><span>()</span> <span>|</span> <span>{</span><span>"</span><span>completed</span><span>"</span><span>:</span> <span>False</span><span>}</span>

    <span>result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>insert_one</span><span>(</span><span>data</span><span>)</span>

    <span>id</span> <span>=</span> <span>str</span><span>(</span><span>result</span><span>.</span><span>inserted_id</span><span>)</span>

    <span>return</span> <span>TodoId</span><span>(</span><span>id</span><span>=</span><span>id</span><span>)</span>

<span>@router.get</span><span>(</span><span>"</span><span>/todo</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_all_todos</span><span>()</span> <span>-></span> <span>list</span><span>[</span><span>Todo</span><span>]:</span>
    <span>"""</span><span> Get all todos </span><span>"""</span>
    <span>result</span> <span>=</span> <span>[]</span>
    <span>async</span> <span>for</span> <span>todo</span> <span>in</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find</span><span>():</span>
        <span>result</span><span>.</span><span>append</span><span>(</span>
            <span>Todo</span><span>(</span>
                <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
                <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
                <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
                <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
            <span>)</span>
        <span>)</span>

    <span>return</span> <span>result</span>

<span>@router.get</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>get_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>Todo</span><span>:</span>
    <span>"""</span><span> Get todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>todo</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>find_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>
    <span>if</span> <span>not</span> <span>todo</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span><span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found</span><span>"</span><span>)</span>

    <span>return</span> <span>Todo</span><span>(</span>
        <span>id</span><span>=</span><span>str</span><span>(</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>_id</span><span>"</span><span>)),</span>
        <span>title</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>title</span><span>"</span><span>),</span>
        <span>description</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>description</span><span>"</span><span>),</span>
        <span>completed</span><span>=</span><span>todo</span><span>.</span><span>get</span><span>(</span><span>"</span><span>completed</span><span>"</span><span>),</span>
    <span>)</span>

<span>@router.patch</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>update_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>,</span> <span>todo_update</span><span>:</span> <span>UpdateTodo</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Update todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>update_data</span> <span>=</span> <span>todo_update</span><span>.</span><span>model_dump</span><span>(</span><span>exclude_unset</span><span>=</span><span>True</span><span>)</span>

    <span>update_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>update_one</span><span>(</span>
        <span>{</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>},</span> <span>{</span><span>"</span><span>$set</span><span>"</span><span>:</span> <span>update_data</span><span>}</span>
    <span>)</span>

    <span>if</span> <span>update_result</span><span>.</span><span>matched_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>

<span>@router.delete</span><span>(</span><span>"</span><span>/todo/{todo_id}</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>delete_one_todo</span><span>(</span><span>todo_id</span><span>:</span> <span>str</span><span>)</span> <span>-></span> <span>SuccessResult</span><span>:</span>
    <span>"""</span><span> Delete todo by ID </span><span>"""</span>
    <span>todo_object_id</span> <span>=</span> <span>ObjectId</span><span>(</span><span>todo_id</span><span>)</span>

    <span>delete_result</span> <span>=</span> <span>await</span> <span>db</span><span>.</span><span>todos</span><span>.</span><span>delete_one</span><span>({</span><span>"</span><span>_id</span><span>"</span><span>:</span> <span>todo_object_id</span><span>})</span>

    <span>if</span> <span>delete_result</span><span>.</span><span>deleted_count</span> <span>==</span> <span>0</span><span>:</span>
        <span>raise</span> <span>HTTPException</span><span>(</span>
            <span>status_code</span><span>=</span><span>status</span><span>.</span><span>HTTP_404_NOT_FOUND</span><span>,</span> <span>detail</span><span>=</span><span>"</span><span>Todo not found.</span><span>"</span>
        <span>)</span>

    <span>return</span> <span>SuccessResult</span><span>(</span><span>success</span><span>=</span><span>True</span><span>)</span>
from bson import ObjectId
from fastapi import APIRouter, HTTPException, status

from app.utils.config import db

from .models import CreateTodo, SuccessResult, Todo, TodoId, UpdateTodo

@router.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """ Create a new todo """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoId(id=id)

@router.get("/todo")
async def get_all_todos() -> list[Todo]:
    """ Get all todos """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get("_id")),
                title=todo.get("title"),
                description=todo.get("description"),
                completed=todo.get("completed"),
            )
        )

    return result

@router.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """ Get todo by ID """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get("_id")),
        title=todo.get("title"),
        description=todo.get("description"),
        completed=todo.get("completed"),
    )

@router.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """ Update todo by ID """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

@router.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """ Delete todo by ID """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Enter fullscreen mode Exit fullscreen mode

Main.py

At this point you main.py should only have app = and middleware.

From todos you will import the whole todo file

Then do app.include_router(todo.router)

<span>import</span> <span>time</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Any</span><span>,</span> <span>Callable</span><span>,</span> <span>TypeVar</span>
<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span>
<span>from</span> <span>fastapi.middleware.cors</span> <span>import</span> <span>CORSMiddleware</span>
<span>from</span> <span>app.routers.todos</span> <span>import</span> <span>todo</span>
<span>app</span> <span>=</span> <span>FastAPI</span><span>(</span>    
        <span>title</span><span>=</span><span>"</span><span>Todo</span><span>"</span><span>,</span>
    <span>description</span><span>=</span><span>"</span><span>What do I need to do?</span><span>"</span><span>,</span>
    <span>version</span><span>=</span><span>"</span><span>1.0.0</span><span>"</span><span>,</span>
    <span>docs_url</span><span>=</span><span>"</span><span>/</span><span>"</span><span>,</span>
<span>)</span>
<span>app</span><span>.</span><span>add_middleware</span><span>(</span>
    <span>CORSMiddleware</span><span>,</span>
    <span>allow_credentials</span><span>=</span><span>True</span><span>,</span>
    <span>allow_methods</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_headers</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_origins</span><span>=</span><span>[</span>
        <span>"</span><span>http://localhost:3000</span><span>"</span><span>,</span>
        <span>"</span><span>http://localhost:8000</span><span>"</span><span>,</span>
    <span>],</span>
<span>)</span>
<span>@app.middleware</span><span>(</span><span>"</span><span>http</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>process_time_log_middleware</span><span>(</span><span>request</span><span>:</span> <span>Request</span><span>,</span> <span>call_next</span><span>:</span> <span>F</span><span>)</span> <span>-></span> <span>Response</span><span>:</span>
    <span>"""</span><span> Add API process time in response headers and log calls </span><span>"""</span>
    <span>start_time</span> <span>=</span> <span>time</span><span>.</span><span>time</span><span>()</span>
    <span>response</span><span>:</span> <span>Response</span> <span>=</span> <span>await</span> <span>call_next</span><span>(</span><span>request</span><span>)</span>
    <span>process_time</span> <span>=</span> <span>str</span><span>(</span><span>round</span><span>(</span><span>time</span><span>.</span><span>time</span><span>()</span> <span>-</span> <span>start_time</span><span>,</span> <span>3</span><span>))</span>
    <span>response</span><span>.</span><span>headers</span><span>[</span><span>"</span><span>X-Process-Time</span><span>"</span><span>]</span> <span>=</span> <span>process_time</span>
    <span>logger</span><span>.</span><span>info</span><span>(</span>
        <span>"</span><span>Method=%s Path=%s StatusCode=%s ProcessTime=%s</span><span>"</span><span>,</span>
        <span>request</span><span>.</span><span>method</span><span>,</span>
        <span>request</span><span>.</span><span>url</span><span>.</span><span>path</span><span>,</span>
        <span>response</span><span>.</span><span>status_code</span><span>,</span>
        <span>process_time</span><span>,</span>
    <span>)</span>
    <span>return</span> <span>response</span>
<span>app</span><span>.</span><span>include_router</span><span>(</span><span>todo</span><span>.</span><span>router</span><span>)</span>
<span>import</span> <span>time</span>
<span>from</span> <span>typing</span> <span>import</span> <span>Any</span><span>,</span> <span>Callable</span><span>,</span> <span>TypeVar</span>

<span>from</span> <span>fastapi</span> <span>import</span> <span>FastAPI</span>
<span>from</span> <span>fastapi.middleware.cors</span> <span>import</span> <span>CORSMiddleware</span>


<span>from</span> <span>app.routers.todos</span> <span>import</span> <span>todo</span>

<span>app</span> <span>=</span> <span>FastAPI</span><span>(</span>    
        <span>title</span><span>=</span><span>"</span><span>Todo</span><span>"</span><span>,</span>
    <span>description</span><span>=</span><span>"</span><span>What do I need to do?</span><span>"</span><span>,</span>
    <span>version</span><span>=</span><span>"</span><span>1.0.0</span><span>"</span><span>,</span>
    <span>docs_url</span><span>=</span><span>"</span><span>/</span><span>"</span><span>,</span>
<span>)</span>

<span>app</span><span>.</span><span>add_middleware</span><span>(</span>
    <span>CORSMiddleware</span><span>,</span>
    <span>allow_credentials</span><span>=</span><span>True</span><span>,</span>
    <span>allow_methods</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_headers</span><span>=</span><span>[</span><span>"</span><span>*</span><span>"</span><span>],</span>
    <span>allow_origins</span><span>=</span><span>[</span>
        <span>"</span><span>http://localhost:3000</span><span>"</span><span>,</span>
        <span>"</span><span>http://localhost:8000</span><span>"</span><span>,</span>
    <span>],</span>
<span>)</span>

<span>@app.middleware</span><span>(</span><span>"</span><span>http</span><span>"</span><span>)</span>
<span>async</span> <span>def</span> <span>process_time_log_middleware</span><span>(</span><span>request</span><span>:</span> <span>Request</span><span>,</span> <span>call_next</span><span>:</span> <span>F</span><span>)</span> <span>-></span> <span>Response</span><span>:</span>
    <span>"""</span><span> Add API process time in response headers and log calls </span><span>"""</span>
    <span>start_time</span> <span>=</span> <span>time</span><span>.</span><span>time</span><span>()</span>
    <span>response</span><span>:</span> <span>Response</span> <span>=</span> <span>await</span> <span>call_next</span><span>(</span><span>request</span><span>)</span>
    <span>process_time</span> <span>=</span> <span>str</span><span>(</span><span>round</span><span>(</span><span>time</span><span>.</span><span>time</span><span>()</span> <span>-</span> <span>start_time</span><span>,</span> <span>3</span><span>))</span>
    <span>response</span><span>.</span><span>headers</span><span>[</span><span>"</span><span>X-Process-Time</span><span>"</span><span>]</span> <span>=</span> <span>process_time</span>

    <span>logger</span><span>.</span><span>info</span><span>(</span>
        <span>"</span><span>Method=%s Path=%s StatusCode=%s ProcessTime=%s</span><span>"</span><span>,</span>
        <span>request</span><span>.</span><span>method</span><span>,</span>
        <span>request</span><span>.</span><span>url</span><span>.</span><span>path</span><span>,</span>
        <span>response</span><span>.</span><span>status_code</span><span>,</span>
        <span>process_time</span><span>,</span>
    <span>)</span>

    <span>return</span> <span>response</span>

<span>app</span><span>.</span><span>include_router</span><span>(</span><span>todo</span><span>.</span><span>router</span><span>)</span>
import time
from typing import Any, Callable, TypeVar

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware


from app.routers.todos import todo

app = FastAPI(    
        title="Todo",
    description="What do I need to do?",
    version="1.0.0",
    docs_url="/",
)

app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_origins=[
        "http://localhost:3000",
        "http://localhost:8000",
    ],
)

@app.middleware("http")
async def process_time_log_middleware(request: Request, call_next: F) -> Response:
    """ Add API process time in response headers and log calls """
    start_time = time.time()
    response: Response = await call_next(request)
    process_time = str(round(time.time() - start_time, 3))
    response.headers["X-Process-Time"] = process_time

    logger.info(
        "Method=%s Path=%s StatusCode=%s ProcessTime=%s",
        request.method,
        request.url.path,
        response.status_code,
        process_time,
    )

    return response

app.include_router(todo.router)

Enter fullscreen mode Exit fullscreen mode

Conclusion

You’ve now built a complete FastAPI application that follows modern Python best practices. The Todo API demonstrates several important concepts:

• Organization using a modular approach with routers and utilities makes the code maintainable, readable and scalable
• Data Validation with Pydantic models ensure type safety and data consistency
• Database Integration with MongoDB
• Error Handling with status codes and error messages improve API usability
• Monitoring with logging middleware which helps track API performance and debug issues
• Frontend Integration with CORS configuration e

In the future I will expand on this project with:

• Adding user authentication
• Creating additional endpoints
• Adding test coverage
• Containerizing with Docker
• Setting up CI/CD pipelines

Remember to keep your dependencies updated and refer to FastAPI’s official documentation for the latest best practices and features.

原文链接：Making a Todo API with FastAPI and MongoDB