```
#!/usr/bin/env python3
# clean_architecture_mandala.py
"""
A tiny, single-file demonstration of Clean Architecture in Python 3.12+.

Tech-stack:
  • dataclasses         – Domain entities
  • pydantic v2         – Input / Output DTOs + validation
  • sqlite3 (std-lib)   – Relational persistence
  • flet                – Reactive multiplatform GUI

Run:
  pip install flet pydantic
  python clean_architecture_mandala.py
"""

###############################################################################
# 1. Domain layer (Enterprise business rules) #################################
###############################################################################
from abc import ABC, abstractmethod
from dataclasses import dataclass, replace
from datetime import datetime, timezone
from enum import Enum, auto


# --- ドメイン例外 ---
class DomainError(Exception): pass


class TitleLengthError(DomainError): pass


# --- Clock インターフェース ---
class Clock(ABC):
    @abstractmethod
    def now(self) -> datetime: ...


# --- Value Object for TaskId ---
@dataclass(frozen=True)
class TaskId:
    value: int

    def __post_init__(self):
        if self.value <= 0:
            raise DomainError("TaskId must be positive")


# --- Value Object for Title ---
@dataclass(frozen=True)
class Title:
    value: str

    def __post_init__(self):
        if not (1 <= len(self.value) <= 100):
            raise TitleLengthError("title must be 1-100 chars")


# --- ステータス列挙 ---
class Status(Enum):
    INCOMPLETE = auto()
    DONE = auto()


# --- Task Entity ---
@dataclass(frozen=True)
class Task:
    id: TaskId
    title: Title
    status: Status
    created_at: datetime

    def toggle(self) -> "Task":
        ns = Status.DONE if self.status == Status.INCOMPLETE else Status.INCOMPLETE
        return replace(self, status=ns)

    @property
    def done(self) -> bool:
        return self.status == Status.DONE


# --- Factory for Task Creation ---
class TaskFactory:
    def __init__(self, clock: Clock):
        self._clock = clock

    def create(self, id_value: int, title_str: str) -> Task:
        task_id = TaskId(id_value)  # TaskIdオブジェクトを生成
        title = Title(title_str)
        return Task(
            id=task_id,
            title=title,
            status=Status.INCOMPLETE,
            created_at=self._clock.now()
        )


###############################################################################
# 2. Use-case layer (Application business rules) ##############################
###############################################################################
from pydantic import BaseModel, Field, ValidationError  # noqa: E402
from typing import Protocol, Self  # noqa: E402


# ─────────────────────────────────────────────────────────────────────────────
# Repository port (Boundary)
#   • Domain エンティティを直接扱い、
#   • 永続化の実装には一切依存しない。
# Infrastructure はこの Protocol を実装して依存を逆転させる。
# ─────────────────────────────────────────────────────────────────────────────
class TaskRepository(Protocol):
    def add(self, task: Task) -> Task: ...

    def list(self) -> list[Task]: ...

    def get(self, task_id: int) -> Task: ...

    def update(self, task: Task) -> Task: ...


# DTO (Input / Output) ────────────────────────────────────────────────────────
class AddTaskIn(BaseModel):
    title: str = Field(min_length=1, max_length=100)


class TaskOut(BaseModel):
    id: int
    title: str
    done: bool

    @classmethod
    def from_entity(cls, t: Task) -> Self:
        # Entity → DTO 変換 (純粋な写像。外部世界のフォーマットを意識しない)
        return cls(
            id=t.id.value,
            title=t.title.value,
            done=t.done
        )


# Use-case 実装 ────────────────────────────────────────────────────────────────
class AddTaskUseCase:

    def __init__(self, repo: TaskRepository, factory: "TaskFactory"):
        self._repo = repo
        self._factory = factory

    def __call__(self, data: AddTaskIn) -> TaskOut:
        # RDB で自動採番するので id はダミー値 1 を入れておく
        draft = self._factory.create(id_value=1, title_str=data.title)
        stored = self._repo.add(draft)  # Repository が採番後の Entity を返す
        return TaskOut.from_entity(stored)


class ListTasksUseCase:
    """Task 一覧取得ユースケース"""

    def __init__(self, repo: TaskRepository):
        self._repo = repo

    def __call__(self) -> list[TaskOut]:
        return [TaskOut.from_entity(t) for t in self._repo.list()]


class ToggleTaskUseCase:
    """Task 完了トグルユースケース"""

    def __init__(self, repo: TaskRepository):
        self._repo = repo

    def __call__(self, task_id: int) -> TaskOut:
        task = self._repo.get(task_id)
        toggled_task = task.toggle()
        return TaskOut.from_entity(self._repo.update(toggled_task))


###############################################################################
# 3. Infrastructure layer (Storage, frameworks etc.) ##########################
###############################################################################
import sqlite3  # noqa: E402
from contextlib import contextmanager  # noqa: E402
from pathlib import Path  # noqa: E402


class SystemClock(Clock):
    def now(self) -> datetime:
        return datetime.now(timezone.utc)


@contextmanager
def _connect(db_path: Path):
    con = sqlite3.connect(db_path)
    try:
        yield con
        if con.in_transaction:  # commit only when write happened
            con.commit()
    finally:
        con.close()


class SQLiteTaskRepository(TaskRepository):
    """
    Domain ↔ SQLite のマッピングを一手に引き受ける Infrastructure 実装。
    テーブル定義は Domain に合わせて `status` を文字列で保存する。
    """

    _DDL = """\
    CREATE TABLE IF NOT EXISTS tasks(
        id         INTEGER PRIMARY KEY AUTOINCREMENT,
        title      TEXT    NOT NULL,
        status     TEXT    NOT NULL CHECK(status IN ('INCOMPLETE','DONE')),
        created_at TEXT    NOT NULL
    )"""

    def __init__(self, db_path: Path = Path("tasks.db")):
        self._db_path = db_path
        self._ensure_schema()

    # ────────────────────────────────────────────────────────── private helpers
    def _ensure_schema(self) -> None:
        with _connect(self._db_path) as con:
            con.execute(self._DDL)

    @staticmethod
    def _row_to_entity(row: tuple) -> Task:
        return Task(
            id=TaskId(row[0]),
            title=Title(row[1]),
            status=Status[row[2]],
            created_at=datetime.fromisoformat(row[3]),
        )

    # ─────────────────────────────────────────────── TaskRepository interface
    def add(self, task: Task) -> Task:
        with _connect(self._db_path) as con:
            cur = con.execute(
                "INSERT INTO tasks(title,status,created_at) VALUES(?,?,?)",
                (task.title.value, task.status.name, task.created_at.isoformat()),
            )
            new_id = cur.lastrowid
        # Task は frozen dataclass なので replace() で新インスタンスを返す
        return replace(task, id=TaskId(new_id))

    def list(self) -> list[Task]:
        with _connect(self._db_path) as con:
            rows = con.execute(
                "SELECT id,title,status,created_at FROM tasks ORDER BY id"
            )
            return [self._row_to_entity(r) for r in rows]

    def get(self, task_id: int) -> Task:
        with _connect(self._db_path) as con:
            row = con.execute(
                "SELECT id,title,status,created_at FROM tasks WHERE id=?",
                (task_id,),
            ).fetchone()
        if row is None:
            raise KeyError(f"Task {task_id} not found")
        return self._row_to_entity(row)

    def update(self, task: Task) -> Task:
        with _connect(self._db_path) as con:
            con.execute(
                "UPDATE tasks SET title=?, status=? WHERE id=?",
                (task.title.value, task.status.name, task.id.value),
            )
        return task


###############################################################################
# 4. Interface adapters (Controllers / View-Models) ###########################
###############################################################################

class TaskController:
    def __init__(
            self,
            add_uc: AddTaskUseCase,
            list_uc: ListTasksUseCase,
            toggle_uc: ToggleTaskUseCase,
    ):
        # D.I. (依存性注入): 完成したUse-caseインスタンスを直接受け取る。
        self.add_uc = add_uc
        self.list_uc = list_uc
        self.toggle_uc = toggle_uc

    def add(self, title: str) -> TaskOut | None:
        """
        タスク追加のUse-Caseを呼び出す。
        UIからの生の入力(title)を、Use-Caseが要求する入力DTO(AddTaskIn)に変換する。
        """
        try:
            # Controllerの責務として、UIからの入力をUse-Case用のDTOに変換する
            input_dto = AddTaskIn(title=title)
            return self.add_uc(input_dto)
        except ValidationError:
            # バリデーションエラーはUIに単純な失敗としてNoneを返す
            return None

    def list(self) -> list[TaskOut]:
        """タスク一覧取得のUse-Caseを呼び出す。"""
        return self.list_uc()

    def toggle(self, task_id: int) -> TaskOut:
        """タスクの状態をトグルするUse-Caseを呼び出す。"""
        return self.toggle_uc(task_id)


###############################################################################
# 5. Presentation layer (Flet UI) #############################################
###############################################################################
import flet as ft  # noqa: E402


def main(page: ft.Page) -> None:
    page.title = "🌀 Clean Architecture Mandala"
    page.horizontal_alignment = ft.CrossAxisAlignment.CENTER
    page.scroll = ft.ScrollMode.ADAPTIVE

    # ─────────────────────────────────────────────────────────────────────────
    # 依存性構築: Presentation が最外層として具体実装を組み立てる
    #             (Infrastructure を生成 → Controller に注入)
    # ─────────────────────────────────────────────────────────────────────────
    # 1. Infrastructure (具象) を生成
    clock = SystemClock()
    repo = SQLiteTaskRepository()

    # 2. Domain (Factory) を生成
    factory = TaskFactory(clock=clock)

    # 3. Use-Cases (具象) を生成し、必要な依存を注入
    add_uc = AddTaskUseCase(repo=repo, factory=factory)
    list_uc = ListTasksUseCase(repo=repo)
    toggle_uc = ToggleTaskUseCase(repo=repo)

    # 4. Controller (Interface Adapter) を生成し、Use-Caseを注入
    ctrl = TaskController(
        add_uc=add_uc,
        list_uc=list_uc,
        toggle_uc=toggle_uc,
    )

    new_task_tf = ft.TextField(
        hint_text="Type a task and hit Enter…",
        width=320,
        on_submit=lambda _: add_task(),
    )
    tasks_column = ft.Column(width=350, spacing=5)

    # ─────────────────────────────────────────────────────────────────────────
    # 以下は View ロジック。Use-case 呼び出し以外は UI のみを扱う。
    # ─────────────────────────────────────────────────────────────────────────
    def add_task() -> None:
        if ctrl.add(new_task_tf.value):
            new_task_tf.value = ""
            render_tasks()

    def on_toggle(e: ft.ControlEvent) -> None:
        ctrl.toggle(int(e.control.data))
        render_tasks()

    def render_tasks() -> None:
        tasks_column.controls = [
            ft.Checkbox(
                label=t.title,
                value=t.done,
                data=str(t.id),
                on_change=on_toggle,
            )
            for t in ctrl.list()
        ]
        page.update()

    page.add(
        ft.Column(
            [
                ft.Text("Clean Architecture – To-Do Example", size=24),
                new_task_tf,
                ft.Divider(height=10, opacity=0),
                tasks_column,
            ],
            horizontal_alignment=ft.CrossAxisAlignment.CENTER,
        )
    )

    render_tasks()


if __name__ == "__main__":
    ft.app(target=main)
```

このコードはどういうもの？