Building a RESTful To-Do API with Clean Architecture in Python

APIs power modern apps, but without good structure they become tangled. In this tutorial, you’ll learn how to create a simple To-Do API in Python—organized with clean architecture patterns for maintainability and testability.

---

What You’ll Build

A minimal To-Do API with endpoints to:

• Create a task
• List all tasks
• Update a task’s status
• Delete a task

All following clean architecture layers so code remains decoupled and easy to evolve.

---

Why Clean Architecture?

Clean architecture separates concerns into layers:

• Entities: Core business objects
• Use Cases: Business rules
• Interface Adapters: Convert data between layers
• Framework & Drivers: External tools (e.g., Flask)

Benefits:

• Testable business logic
• Easier to swap frameworks or databases
• Clear dependency rule: outer layers depend on inner ones

---

Project Structure

📂 todo-api
 ├── 📂 entities
 │    └── 📄 task
 ├── 📂 use_cases
 │    └── 📄 task_manager
 ├── 📂 adapters
 │    ├── 📄 repository
 │    └── 📄 controller
 ├── 📂 infrastructure
 │    └── 📄 flask_app
 └── 📂 tests
      └── 📄 test_use_cases

---

Step 1: Define Entities

# entities/task.py
from dataclasses import dataclass
from uuid import uuid4

@dataclass
class Task:
    id: str
    title: str
    completed: bool = False

    @staticmethod
    def create(title: str) -> "Task":
        return Task(id=str(uuid4()), title=title)

• Task holds data and invariants.
• create() ensures every task gets a unique ID.

---

Step 2: Use Cases (Business Logic)

# use_cases/task_manager.py
from typing import List
from entities.task import Task
from adapters.repository import TaskRepository

class TaskManager:
    def __init__(self, repo: TaskRepository):
        self.repo = repo

    def add_task(self, title: str) -> Task:
        task = Task.create(title)
        return self.repo.save(task)

    def list_tasks(self) -> List[Task]:
        return self.repo.get_all()

    def mark_completed(self, task_id: str) -> Task:
        task = self.repo.get_by_id(task_id)
        task.completed = True
        return self.repo.save(task)

    def delete_task(self, task_id: str) -> None:
        self.repo.delete(task_id)

• Business rules live here—no Flask or database code.

---

Step 3: Interface Adapters (Controllers & Repos)

# adapters/repository.py
from entities.task import Task

class TaskRepository:
    def __init__(self):
        self.storage: dict[str, Task] = {}

    def save(self, task: Task) -> Task:
        self.storage[task.id] = task
        return task

    def get_all(self) -> list[Task]:
        return list(self.storage.values())

    def get_by_id(self, task_id: str) -> Task:
        return self.storage[task_id]

    def delete(self, task_id: str) -> None:
        del self.storage[task_id]

# adapters/controller.py
from flask import request, jsonify, Blueprint
from use_cases.task_manager import TaskManager
from adapters.repository import TaskRepository

bp = Blueprint('tasks', __name__)
repo = TaskRepository()
manager = TaskManager(repo)

@bp.route('/tasks', methods=['POST'])
def create_task():
    data = request.json
    task = manager.add_task(data['title'])
    return jsonify(vars(task)), 201

@bp.route('/tasks', methods=['GET'])
def list_tasks():
    tasks = manager.list_tasks()
    return jsonify([vars(t) for t in tasks])

@bp.route('/tasks/<id>/complete', methods=['PATCH'])
def complete_task(id):
    task = manager.mark_completed(id)
    return jsonify(vars(task))

@bp.route('/tasks/<id>', methods=['DELETE'])
def delete_task(id):
    manager.delete_task(id)
    return '', 204

---

Step 4: Framework & Drivers (Flask)

# infrastructure/flask_app.py
from flask import Flask
from adapters.controller import bp

def create_app():
    app = Flask(__name__)
    app.register_blueprint(bp, url_prefix='/api')
    return app

if __name__ == '__main__':
    app = create_app()
    app.run(debug=True)

• Runs on http://localhost:5000/api/tasks

Step 5: Testing

# tests/test_use_cases.py
import pytest
from use_cases.task_manager import TaskManager
from adapters.repository import TaskRepository

@pytest.fixture
def manager():
    return TaskManager(TaskRepository())

def test_add_and_list(manager):
    manager.add_task("Write blog post")
    tasks = manager.list_tasks()
    assert len(tasks) == 1
    assert tasks[0].title == "Write blog post"

• Focus on use cases—no need to spin up Flask.

Next Steps

• Persist to a real database (SQLAlchemy adapter).
• Add authentication (JWT middleware).
• Expand entities (due dates, priorities).