Skip to content

#231: Split a FastAPI Application Into Manageable Parts

As we saw with the OAuth2 and JWT example, our main.py file can grow quickly. If we keep adding features, the file will be unmaintainable in no time. For Flask we used Blueprint to split up the application , for FastAPI we can use APIRouter to do the same.

Our starting point

To explain the concept of routers, we best start with something smaller. We return to our minimalistic to-do application and continue where we left it . Our main.py currently looks like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
from typing import List
from fastapi import Depends, FastAPI, HTTPException, Request, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from .models.todo import *
from .data.datastore import DataStore

app = FastAPI()

db = DataStore()

async def filter_parameters(q: str | None = None, 
                            include_done: bool = True, 
                            due_before: date = date.today() + timedelta(days=365)):
    return {"q": q, "include_done": include_done, "due_before": due_before }


@app.get("/", include_in_schema=False)
async def main():
    return {'message':'The minimalistic ToDo API'}


@app.get("/api/todo")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)]) -> List[TaskOutput]:
    result = db.all()

    if not filter["include_done"]:
        result = [item for item in result if item.done == False ]

    result = [item for item in result if item.due_date <= filter["due_before"] ]

    return result

@app.post("/api/todo", status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskInput, request: Request) -> TaskOutput:
    result = db.add(task)
    headers = {"Location": f"{request.base_url}api/todo/{result.id}"}
    return JSONResponse(content=jsonable_encoder(result), 
                        status_code=status.HTTP_201_CREATED,
                        headers=headers)


@app.get("/api/todo/{id}")
async def show_task(id: int) -> TaskOutput:
    result = db.get(id)

    if result:
        return result
    else:
        raise HTTPException(status_code=404, detail="Task not found")



@app.put("/api/todo/{id}")
async def update_task(id: int, task: TaskInput) -> TaskOutput:
    try:
        result = db.update(id, task)
        return result
    except ValueError:
        raise HTTPException(status_code=404, detail="Task not found")


@app.delete("/api/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(id: int) -> None:
    db.delete(id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Create the router

To move our existing endpoints into a new router, we can follow this checklist:

  1. Create a new folder routers.
  2. Create the empty __init__.py inside routers.
  3. Create a todo.py inside routers.
  4. Copy your existing endpoints (all except main()) to routers/todo.py.
  5. Create a router object with router = APIRouter()
  6. Replace @app with @router

Our extracted endpoints file now looks like this:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
from datetime import date, timedelta
from typing import Annotated, List

from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

from ..models.todo import TaskOutput, TaskInput
from ..data.datastore import DataStore

router = APIRouter()

db = DataStore()

async def filter_parameters(q: str | None = None, 
                            include_done: bool = True, 
                            due_before: date = date.today() + timedelta(days=365)):
    return {"q": q, "include_done": include_done, "due_before": due_before }


@router.get("/api/todo")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)]) -> List[TaskOutput]:
    result = db.all()

    if not filter["include_done"]:
        result = [item for item in result if item.done == False ]

    result = [item for item in result if item.due_date <= filter["due_before"] ]

    return result


@router.post("/api/todo", status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskInput, request: Request) -> TaskOutput:
    result = db.add(task)
    headers = {"Location": f"{request.base_url}api/todo/{result.id}"}
    return JSONResponse(content=jsonable_encoder(result), 
                        status_code=status.HTTP_201_CREATED,
                        headers=headers)


@router.get("/api/todo/{id}")
async def show_task(id: int) -> TaskOutput:
    result = db.get(id)

    if result:
        return result
    else:
        raise HTTPException(status_code=404, detail="Task not found")


@router.put("/api/todo/{id}")
async def update_task(id: int, task: TaskInput) -> TaskOutput:
    try:
        result = db.update(id, task)
        return result
    except ValueError:
        raise HTTPException(status_code=404, detail="Task not found")


@router.delete("/api/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(id: int) -> None:
    db.delete(id)
    return Response(status_code=status.HTTP_204_NO_CONTENT)

Include the router in main.py

In the main.py file, we can remove the extracted endpoints, get rid of all no longer needed imports and tell FastAPI that we want to include our todo router:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from fastapi import Depends, FastAPI
from .routers import todo


app = FastAPI()
app.include_router(todo.router)


@app.get("/", include_in_schema=False)
async def main():
    return {'message':'The minimalistic ToDo API'}

Run the tests

We only moved code around inside our application. Therefore, all our tests should still work, and we better check if this is indeed the case:

pytest -q .\minimal_todo\tests\

.......................... [100%] 26 passed in 0.67s
 

Optimise the Router

We currently have /api/todo in the route of all our endpoints. We can add this part as a prefix to the include_router() method in main.py and remove it from the endpoints in our router:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from fastapi import FastAPI
from .routers import todo


app = FastAPI()
app.include_router(todo.router, prefix="/api/todo")


@app.get("/", include_in_schema=False)
async def main():
    return {'message':'The minimalistic ToDo API'}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
...
@router.get("/")
async def show_all_tasks(filter: Annotated[dict, Depends(filter_parameters)]) -> List[TaskOutput]:
...

@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_task(task: TaskInput, request: Request) -> TaskOutput:
...

@router.get("/{id}")
async def show_task(id: int) -> TaskOutput:
...    

@router.put("/{id}")
async def update_task(id: int, task: TaskInput) -> TaskOutput:
...    

@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_task(id: int) -> None:
...

We can run our tests to check that everything still works.

This allows us to change the route for our endpoints at a single place, should that be something we want to do in the future.

Attention: If you set the prefix in include_router() and in APIRouter(), the two prefixes get combined.

Next

With APIRouter we not only can separate endpoints by topic, but we can also add endpoints from other packages. That can be helpful for large applications or when we want to use a library to handle all the authentication for us. Next week we extend the to-do application and add a database.