Controller

A controller receives a request, then calls a use case, before finally returning a response.

The controller (adapter layer) is responsible for validating and transforming requests into an understandable format for the use cases (application layer). The format is defined inside the use cases by the request and response models. The controller takes the user input (the request), converts it into the request model defined by the use case and passes the request model to the use case, and at the end return the response model.

from fastapi import APIRouter, Depends

from authentication.authentication import auth_with_jwt
from authentication.models import User
from features.todo.repository.todo_repository import get_todo_repository
from features.todo.repository.todo_repository_interface import TodoRepositoryInterface
from features.todo.use_cases.add_todo import (
    AddTodoRequest,
    AddTodoResponse,
    add_todo_use_case,
)
from features.todo.use_cases.delete_todo_by_id import (
    DeleteTodoByIdResponse,
    delete_todo_use_case,
)
from features.todo.use_cases.get_todo_all import GetTodoAllResponse, get_todo_all_use_case
from features.todo.use_cases.get_todo_by_id import (
    GetTodoByIdResponse,
    get_todo_by_id_use_case,
)
from features.todo.use_cases.update_todo import (
    UpdateTodoRequest,
    UpdateTodoResponse,
    update_todo_use_case,
)

router = APIRouter(tags=["todos"], prefix="/todos")


@router.post("", operation_id="create")
def add_todo(
    data: AddTodoRequest,
    user: User = Depends(auth_with_jwt),
    todo_repository: TodoRepositoryInterface = Depends(get_todo_repository),
) -> AddTodoResponse:
    return add_todo_use_case(data=data, user_id=user.user_id, todo_repository=todo_repository)


@router.get("/{id}", operation_id="get_by_id")
def get_todo_by_id(
    id: str,
    user: User = Depends(auth_with_jwt),
    todo_repository: TodoRepositoryInterface = Depends(get_todo_repository),
) -> GetTodoByIdResponse:
    return get_todo_by_id_use_case(id=id, user_id=user.user_id, todo_repository=todo_repository)


@router.delete("/{id}", operation_id="delete_by_id")
def delete_todo_by_id(
    id: str,
    user: User = Depends(auth_with_jwt),
    todo_repository: TodoRepositoryInterface = Depends(get_todo_repository),
) -> DeleteTodoByIdResponse:
    return delete_todo_use_case(id=id, user_id=user.user_id, todo_repository=todo_repository)


@router.get("", operation_id="get_all")
def get_todo_all(
    user: User = Depends(auth_with_jwt), todo_repository: TodoRepositoryInterface = Depends(get_todo_repository)
) -> list[GetTodoAllResponse]:
    return get_todo_all_use_case(user_id=user.user_id, todo_repository=todo_repository)  # type: ignore


@router.put("/{id}", operation_id="update_by_id")
def update_todo(
    id: str,
    data: UpdateTodoRequest,
    user: User = Depends(auth_with_jwt),
    todo_repository: TodoRepositoryInterface = Depends(get_todo_repository),
) -> UpdateTodoResponse:
    return update_todo_use_case(id=id, data=data, user_id=user.user_id, todo_repository=todo_repository)

Required
- The controller needs to be decorated with the create_response decorator, which handles exceptions and returns a unified response type.
- The controller needs to have set the response_model and request_model, that is used to generate API documentation and used for validation.
Optional
- Add repository interface to handle communication to external services such as databases and inject the repository implementations to the controller endpoint and pass the injected repository implementations to the use case.

note

FastAPI is built around the OpenAPI Specification (formerly known as swagger) standards. In FastAPI, by coding your endpoints, you are automatically writing your API documentation. FastAPI maps your endpoint details to a JSON Schema document. Under the hood, FastAPI uses Pydantic for data validation. With Pydantic along with type hints, you get a nice editor experience with autocompletion.

Testing controllers

Use the test_client fixture to populate the database with test data and test_app fixture to perform REST API calls.

import pytest
from starlette.status import (
    HTTP_200_OK,
    HTTP_404_NOT_FOUND,
    HTTP_422_UNPROCESSABLE_ENTITY,
)
from starlette.testclient import TestClient

from data_providers.clients.client_interface import ClientInterface


class TestTodo:
    @pytest.fixture(autouse=True)
    def setup_database(self, test_client: ClientInterface):
        test_client.insert_many(
            [
                {"_id": "1", "id": "1", "title": "title 1", "user_id": "nologin"},
                {"_id": "2", "id": "2", "title": "title 2", "user_id": "nologin"},
            ]
        )

    def test_get_todo_all(self, test_app: TestClient):
        response = test_app.get("/todos")
        items = response.json()

        assert response.status_code == HTTP_200_OK
        assert len(items) == 2
        assert items[0]["id"] == "1"
        assert items[0]["title"] == "title 1"
        assert items[1]["id"] == "2"
        assert items[1]["title"] == "title 2"

    def test_get_todo_by_id(self, test_app: TestClient):
        response = test_app.get("/todos/1")

        assert response.status_code == HTTP_200_OK
        assert response.json()["id"] == "1"
        assert response.json()["title"] == "title 1"

    def test_get_todo_should_return_not_found(self, test_app: TestClient):
        response = test_app.get("/todos/unknown")
        assert response.status_code == HTTP_404_NOT_FOUND

    def test_add_todo(self, test_app: TestClient):
        response = test_app.post("/todos", json={"title": "title 3"})
        item = response.json()

        assert response.status_code == HTTP_200_OK
        assert item["title"] == "title 3"

    def test_add_todo_should_return_unprocessable_when_invalid_entity(self, test_app: TestClient):
        response = test_app.post("/todos", json=None)

        assert response.status_code == HTTP_422_UNPROCESSABLE_ENTITY

    def test_update_todo(self, test_app):
        response = test_app.put("/todos/1", json={"title": "title 1 updated", "is_completed": False})

        assert response.status_code == HTTP_200_OK
        assert response.json()["success"]

    def test_update_todo_should_return_not_found(self, test_app):
        response = test_app.put("/todos/unknown", json={"title": "something", "is_completed": False})
        assert response.status_code == HTTP_404_NOT_FOUND

    def test_update_todo_should_return_unprocessable_when_invalid_entity(self, test_app: TestClient):
        response = test_app.put("/todos/1", json={"title": ""})

        assert response.status_code == HTTP_422_UNPROCESSABLE_ENTITY

    def test_delete_todo(self, test_app: TestClient):
        response = test_app.delete("/todos/1")

        assert response.status_code == HTTP_200_OK
        assert response.json()["success"]

    def test_delete_todo_should_return_not_found(self, test_app: TestClient):
        response = test_app.delete("/todos/unknown")
        assert response.status_code == HTTP_404_NOT_FOUND

note

Mark it as integration test.

Controller

Testing controllers​

Testing controllers