FastAPI, Peewee, and Building Blocks

With FastAPI and Peewee, it becomes quite simple to implement the tactical patterns required for this project.

1. Entity

Entities represent domain objects that have a unique identity. In the context of Peewee, entities are defined as models that connect to a table in the database.

Example:

from peewee import Model, CharField, SqliteDatabase

# Database connection
db = SqliteDatabase('users.db')

class User(Model):
    name = CharField()
    email = CharField(unique=True)

    class Meta:
        database = db  # Defines the database associated with the model

2. Repository

The Repository acts as an intermediary between the application and the data persistence layer. It provides methods for manipulating entities.

Example:

class UserRepository:
    @staticmethod
    def create_user(name: str, email: str) -> User:
        user = User.create(name=name, email=email)
        return user

    @staticmethod
    def get_user(user_id: int) -> User:
        return User.get_or_none(User.id == user_id)

    @staticmethod
    def get_all_users() -> list:
        return list(User.select())

    @staticmethod
    def delete_user(user_id: int) -> None:
        user = User.get_or_none(User.id == user_id)
        if user:
            user.delete_instance()

3. Service

The Service encapsulates the application's business logic, orchestrating the interactions between entities and repositories.

Example:

class UserService:
    def __init__(self, user_repository: UserRepository):
        self.user_repository = user_repository

    def create_user(self, name: str, email: str) -> User:
        return self.user_repository.create_user(name, email)

    def get_user(self, user_id: int) -> User:
        return self.user_repository.get_user(user_id)

    def get_all_users(self) -> list:
        return self.user_repository.get_all_users()

    def delete_user(self, user_id: int) -> None:
        self.user_repository.delete_user(user_id)

4. Controller

The Controller handles user requests, coordinating the interaction between the user interface and the services. FastAPI makes it easy to create APIs with routes and data handling.

Example:

from fastapi import FastAPI, HTTPException

app = FastAPI()
user_repository = UserRepository()
user_service = UserService(user_repository)

@app.post("/users/")
def create_user(name: str, email: str):
    user = user_service.create_user(name, email)
    return {"id": user.id, "name": user.name, "email": user.email}

@app.get("/users/{user_id}")
def get_user(user_id: int):
    user = user_service.get_user(user_id)
    if user:
        return {"id": user.id, "name": user.name, "email": user.email}
    raise HTTPException(status_code=404, detail="User not found")

@app.get("/users/")
def get_all_users():
    users = user_service.get_all_users()
    return [{"id": user.id, "name": user.name, "email": user.email} for user in users]

@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    user_service.delete_user(user_id)
    return {"message": "User deleted successfully"}

5. TDD

We will use the TestClient provided by FastAPI. This approach allows for integrated testing of FastAPI endpoints, validating the application's functionality without the need for mocks, and ensuring that the system's behavior is as expected.

import pytest
from fastapi.testclient import TestClient
from your_module import app, UserRepository, UserService, User  # Adjust the import as needed

# TestClient initialization
client = TestClient(app)

# Assuming you have appropriate UserRepository and UserService
@pytest.fixture(scope="module")
def setup_users():
    # Cleanup and initial setup of the user repository
    user_repository = UserRepository()
    user_service = UserService(user_repository)

    # Adding users for testing
    user_service.create_user("Alice", "alice@example.com")
    user_service.create_user("Bob", "bob@example.com")

    yield  # Allows the tests to run

    # Cleanup after tests
    user_repository.clear()  # Assume you have a method to clear the data

def test_create_user():
    response = client.post("/users/", json={"name": "Charlie", "email": "charlie@example.com"})
    assert response.status_code == 200
    data = response.json()
    assert "id" in data
    assert data["name"] == "Charlie"
    assert data["email"] == "charlie@example.com"

def test_get_user(setup_users):
    response = client.get("/users/1")  # Assumes the first user's ID is 1
    assert response.status_code == 200
    data = response.json()
    assert data["name"] == "Alice"
    assert data["email"] == "alice@example.com"

def test_get_user_not_found():
    response = client.get("/users/999")  # ID that does not exist
    assert response.status_code == 404
    assert response.json() == {"detail": "User not found"}

def test_get_all_users(setup_users):
    response = client.get("/users/")
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 2  # Expects 2 users created in the setup
    assert data[0]["name"] == "Alice"
    assert data[1]["name"] == "Bob"

def test_delete_user(setup_users):
    response = client.delete("/users/1")  # Delete the user with ID 1
    assert response.status_code == 200
    assert response.json() == {"message": "User deleted successfully"}

    # Verify if the user was actually deleted
    response = client.get("/users/1")
    assert response.status_code == 404