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.
This post is part of my journey to learn Python. You find the code for this post in my PythonFriday repository on GitHub.
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:
- Create a new folder routers.
- Create the empty __init__.py inside routers.
- Create a todo.py inside routers.
- Copy your existing endpoints (all except main()) to routers/todo.py.
- Create a router object with router = APIRouter()
- 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:
|
1 |
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.
2 thoughts on “Python Friday #231: Split a FastAPI Application Into Manageable Parts”