feat: initial commit for TodoFlow - Conteneurised API and Glassmorphic Desktop Widget

This commit is contained in:
Michael committed 2026-05-26 17:42:14 +02:00
commit a996f0694b
23 files changed
+1897

No files matched your search

+35
View File
@@ -0,0 +1,35 @@
# --- Stage 1: Build dependencies ---
FROM python:3.11-slim AS builder
WORKDIR /app
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
&& rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt
# --- Stage 2: Final Minimal Image ---
FROM python:3.11-slim AS runner
WORKDIR /app
# Copie des dépendances compilées depuis builder
COPY --from=builder /root/.local /root/.local
COPY . /app
ENV PATH=/root/.local/bin:$PATH
ENV PYTHONUNBUFFERED=1
# Création d'un répertoire pour la base de données SQLite persistante
RUN mkdir -p /app/data
# Création d'un utilisateur système non-privilégié pour des raisons de sécurité
RUN useradd -u 8888 appuser && chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
# Lancement d'Uvicorn
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
+132
View File
@@ -0,0 +1,132 @@
# 🤖 Guide d'Intégration pour Agent IA - TodoFlow API
Ce guide explique comment un **Agent d'Intelligence Artificielle** (comme un assistant autonome, un agent de codage ou un script LLM) peut se connecter, comprendre et piloter la Todo List centralisée de l'utilisateur.
---
## 🔑 1. Authentification & Sécurité
Pour toutes les requêtes HTTP, l'Agent IA doit s'authentifier en fournissant la clé API configurée dans l'en-tête `X-API-Key`.
### En-tête HTTP Requis :
```http
X-API-Key: ai-agent-token-secure-101
```
*Note : La clé par défaut pour l'agent IA dans la configuration de production est `ai-agent-token-secure-101`, mais elle peut être personnalisée via la variable d'environnement `API_KEYS_RAW` du conteneur Docker.*
---
## 📐 2. Découverte Automatique (OpenAPI / Swagger)
L'API TodoFlow a été conçue avec **FastAPI**, ce qui signifie qu'elle génère dynamiquement une spécification OpenAPI standard au format JSON.
### Endpoints de Découverte :
* **Schéma OpenAPI JSON :** `http://<IP_SERVEUR>:8000/openapi.json`
*(Idéal pour qu'un LLM lise et comprenne la structure des données et des routes de manière autonome)*
* **Documentation Swagger Interactive :** `http://<IP_SERVEUR>:8000/docs`
---
## 📋 3. Spécifications des Opérations (CRUD)
### A. Récupérer toutes les tâches (GET `/api/v1/tasks`)
Permet à l'Agent IA de lire la liste de tâches actuelle. Il peut filtrer les tâches par statut.
* **Query Parameters (Optionnel) :**
* `status` : Peut être `"A faire"`, `"En cours"` ou `"Termine"`.
* **Réponse (200 OK) :** Une liste d'objets au format suivant :
```json
[
{
"id": "e1a77ff0-2c32-4f37-baef-e68b903f806e",
"title": "Acheter du pain",
"description": null,
"status": "A faire",
"created_at": "2026-05-26T14:15:30Z",
"updated_at": "2026-05-26T14:15:30Z"
}
]
```
### B. Créer une nouvelle tâche (POST `/api/v1/tasks`)
Permet à l'Agent IA d'ajouter un nouvel objectif pour l'utilisateur.
* **Request Body (JSON) :**
```json
{
"title": "Préparer la réunion de projet",
"description": "Rédiger l'ordre du jour et préparer les diapositives",
"status": "A faire"
}
```
* **Réponse (201 Created) :** Renvoie l'objet créé avec son UUID unique généré par le serveur et son horodatage de création (`created_at`).
### C. Mettre à jour une tâche (PUT `/api/v1/tasks/{id}`)
Permet à l'Agent IA de marquer une tâche comme **"En cours"** ou **"Termine"**, ou d'en modifier le titre/description.
* **Request Body (JSON - Mise à jour partielle acceptée) :**
```json
{
"status": "En cours"
}
```
* **Réponse (200 OK) :** Renvoie l'objet mis à jour avec son nouvel horodatage `updated_at`.
### D. Supprimer une tâche (DELETE `/api/v1/tasks/{id}`)
Permet à l'Agent IA de nettoyer la liste en supprimant une tâche.
* **Réponse (204 No Content) :** Succès de la suppression (aucun corps de réponse).
---
## 🐍 4. Exemple d'implémentation en Python pour l'Agent IA
Voici un script simple montrant comment un Agent IA peut interagir avec l'API en utilisant la bibliothèque `requests` :
```python
import requests
API_URL = "http://localhost:8000/api/v1"
API_KEY = "ai-agent-token-secure-101"
headers = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
# 1. Lire toutes les tâches actives
response = requests.get(f"{API_URL}/tasks", headers=headers)
if response.status_code == 200:
tasks = response.json()
print(f"L'utilisateur a {len(tasks)} tâches au total.")
# 2. Créer une nouvelle tâche pour l'utilisateur
new_task = {
"title": "Analyser les performances de la stack",
"description": "Tâche générée automatiquement par l'Agent IA",
"status": "A faire"
}
create_resp = requests.post(f"{API_URL}/tasks", json=new_task, headers=headers)
if create_resp.status_code == 201:
created_task = create_resp.json()
task_id = created_task["id"]
print(f"Tâche créée par l'IA avec succès ! ID: {task_id}")
# 3. Passer cette tâche "En cours"
update_resp = requests.put(
f"{API_URL}/tasks/{task_id}",
json={"status": "En cours"},
headers=headers
)
if update_resp.status_code == 200:
print("La tâche a été marquée 'En cours' par l'IA.")
```
---
## 💡 Conseils pour le Prompt System de l'Agent IA
Si vous intégrez cette API à un LLM (comme GPT-4 ou Gemini), vous pouvez ajouter cette consigne dans son **System Prompt** :
> *"Tu es un Agent IA assistant personnel. Tu as accès à l'API de gestion des tâches de l'utilisateur. Pour piloter sa liste, tu dois faire des appels REST sur l'API en utilisant l'en-tête `X-API-Key: ai-agent-token-secure-101`. Découvre les spécifications exactes de l'API en interrogeant `/openapi.json`. Assure-toi de respecter les statuts valides : 'A faire', 'En cours', 'Termine'."*
+16
View File
@@ -0,0 +1,16 @@
from typing import List
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
DATABASE_URL: str = "sqlite:///./todo.db"
# Clés API séparées par des virgules dans l'environnement, stockées comme liste
API_KEYS_RAW: str = "widget-token-secure-789,ai-agent-token-secure-101"
@property
def api_keys(self) -> List[str]:
return [key.strip() for key in self.API_KEYS_RAW.split(",") if key.strip()]
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
settings = Settings()
+18
View File
@@ -0,0 +1,18 @@
from sqlmodel import SQLModel, create_engine, Session
from app.config import settings
# Configuration de SQLite pour supporter le multi-threading de FastAPI
connect_args = {}
if settings.DATABASE_URL.startswith("sqlite"):
connect_args = {"check_same_thread": False}
engine = create_engine(settings.DATABASE_URL, echo=False, connect_args=connect_args)
def create_db_and_tables():
# Crée les tables si elles n'existent pas
SQLModel.metadata.create_all(engine)
def get_session():
# Fournit une session de base de données à chaque requête API
with Session(engine) as session:
yield session
+203
View File
@@ -0,0 +1,203 @@
from contextlib import asynccontextmanager
from datetime import datetime
from uuid import UUID
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Depends, Security, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.security.api_key import APIKeyHeader
from sqlmodel import Session, select
from app.config import settings
from app.database import create_db_and_tables, get_session
from app.models import Task, TaskCreate, TaskUpdate, TaskResponse, TaskStatus
# Définition du header attendu pour la clé API
API_KEY_NAME = "X-API-Key"
api_key_header = APIKeyHeader(name=API_KEY_NAME, auto_error=False)
# Dépendance pour vérifier la clé API
def verify_api_key(api_key: str = Depends(api_key_header)):
# L'en-tête est obligatoire pour sécuriser l'API des accès non autorisés
if not api_key:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="La clé API est manquante dans l'en-tête X-API-Key."
)
if api_key not in settings.api_keys:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Clé API invalide ou non autorisée."
)
return api_key
# Cycle de vie de l'application (Lifespan modern FastAPI)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Création des tables à l'initialisation du serveur
create_db_and_tables()
yield
app = FastAPI(
title="TodoFlow API",
description="API centralisée pour la Todo list, connectée au widget de bureau et pilotable par Agent IA.",
version="1.0.0",
lifespan=lifespan
)
# Configuration CORS pour permettre au widget (Tauri) de requêter le serveur librement
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Tauri utilise des protocoles personnalisés ou localhost
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# --- ENDPOINTS API ---
@app.get(
"/api/v1/tasks",
response_model=List[TaskResponse],
dependencies=[Depends(verify_api_key)],
summary="Récupérer toutes les tâches"
)
def read_tasks(
status: Optional[TaskStatus] = None,
db: Session = Depends(get_session)
):
"""
Récupère la liste de toutes les tâches stockées.
Permet de filtrer optionnellement par statut (A faire, En cours, Termine).
"""
try:
statement = select(Task)
if status:
statement = statement.where(Task.status == status)
# Tri automatique : d'abord les tâches "En cours", puis "À faire", et enfin les "Terminé".
# En second critère, on trie par date de création décroissante.
tasks = db.exec(statement).all()
# Tri personnalisé côté serveur pour optimiser l'affichage du widget
status_priority = {TaskStatus.IN_PROGRESS: 0, TaskStatus.TODO: 1, TaskStatus.DONE: 2}
tasks.sort(key=lambda t: (status_priority.get(t.status, 9), t.created_at), reverse=False)
return tasks
except Exception as e:
# Journaliser et renvoyer une erreur interne propre
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erreur lors de la récupération des tâches : {str(e)}"
)
@app.get(
"/api/v1/tasks/{task_id}",
response_model=TaskResponse,
dependencies=[Depends(verify_api_key)],
summary="Récupérer les détails d'une tâche"
)
def read_task(task_id: UUID, db: Session = Depends(get_session)):
"""
Récupère une tâche spécifique par son UUID.
Renvoie une erreur 404 si la tâche n'existe pas.
"""
task = db.get(Task, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tâche introuvable."
)
return task
@app.post(
"/api/v1/tasks",
response_model=TaskResponse,
status_code=status.HTTP_201_CREATED,
dependencies=[Depends(verify_api_key)],
summary="Créer une nouvelle tâche"
)
def create_task(task_create: TaskCreate, db: Session = Depends(get_session)):
"""
Crée une nouvelle tâche avec une date de création automatique et un UUID unique.
"""
try:
new_task = Task.model_validate(task_create)
db.add(new_task)
db.commit()
db.refresh(new_task)
return new_task
except Exception as e:
db.rollback()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Impossible de créer la tâche : {str(e)}"
)
@app.put(
"/api/v1/tasks/{task_id}",
response_model=TaskResponse,
dependencies=[Depends(verify_api_key)],
summary="Mettre à jour une tâche existante"
)
def update_task(
task_id: UUID,
task_update: TaskUpdate,
db: Session = Depends(get_session)
):
"""
Met à jour partiellement ou totalement une tâche.
Met automatiquement à jour le champ updated_at.
"""
task = db.get(Task, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tâche introuvable pour mise à jour."
)
try:
# Extraction des champs soumis pour une mise à jour partielle
update_data = task_update.model_dump(exclude_unset=True)
for key, value in update_data.items():
setattr(task, key, value)
# Mise à jour systématique de l'horodatage de modification
task.updated_at = datetime.utcnow()
db.add(task)
db.commit()
db.refresh(task)
return task
except Exception as e:
db.rollback()
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Erreur lors de la mise à jour de la tâche : {str(e)}"
)
@app.delete(
"/api/v1/tasks/{task_id}",
status_code=status.HTTP_204_NO_CONTENT,
dependencies=[Depends(verify_api_key)],
summary="Supprimer une tâche"
)
def delete_task(task_id: UUID, db: Session = Depends(get_session)):
"""
Supprime définitivement une tâche par son UUID.
"""
task = db.get(Task, task_id)
if not task:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Tâche introuvable pour suppression."
)
try:
db.delete(task)
db.commit()
return None
except Exception as e:
db.rollback()
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Erreur lors de la suppression de la tâche : {str(e)}"
)
+35
View File
@@ -0,0 +1,35 @@
from datetime import datetime
from enum import Enum
from typing import Optional
from uuid import UUID, uuid4
from sqlmodel import SQLModel, Field
class TaskStatus(str, Enum):
TODO = "A faire"
IN_PROGRESS = "En cours"
DONE = "Termine"
class TaskBase(SQLModel):
title: str = Field(index=True, min_length=1, max_length=100)
description: Optional[str] = Field(default=None)
status: TaskStatus = Field(default=TaskStatus.TODO)
class TaskCreate(TaskBase):
pass
class TaskUpdate(SQLModel):
title: Optional[str] = Field(default=None, min_length=1, max_length=100)
description: Optional[str] = None
status: Optional[TaskStatus] = None
class TaskResponse(TaskBase):
id: UUID
created_at: datetime
updated_at: datetime
class Task(TaskBase, table=True):
__tablename__ = "tasks"
id: UUID = Field(default_factory=uuid4, primary_key=True, index=True)
created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
updated_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
+25
View File
@@ -0,0 +1,25 @@
version: '3.8'
services:
todo-api:
build:
context: .
dockerfile: Dockerfile
container_name: todoflow-api-server
restart: always
ports:
- "8000:8000"
volumes:
- todoflow_db_volume:/app/data
environment:
- DATABASE_URL=sqlite:///app/data/todo.db
- API_KEYS_RAW=widget-token-secure-789,ai-agent-token-secure-101
- ENV=production
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
volumes:
todoflow_db_volume:
+5
View File
@@ -0,0 +1,5 @@
fastapi>=0.100.0
uvicorn[standard]>=0.22.0
sqlmodel>=0.0.8
pydantic-settings>=2.0.0
pydantic>=2.0.0
+109
View File
@@ -0,0 +1,109 @@
import pytest
from uuid import UUID
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.pool import StaticPool
from app.main import app, verify_api_key
from app.database import get_session
from app.models import Task, TaskStatus
# Clé API valide pour les tests
VALID_API_KEY = "test-token-123"
# Montage d'une base de données SQLite en mémoire pour les tests unitaires
@pytest.fixture(name="session")
def session_fixture():
engine = create_engine(
"sqlite://",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
SQLModel.metadata.create_all(engine)
with Session(engine) as session:
yield session
@pytest.fixture(name="client")
def client_fixture(session: Session):
# Remplacement du générateur de session par la session de test en mémoire
def get_session_override():
return session
app.dependency_overrides[get_session] = get_session_override
# Remplacement des clés API valides par notre clé de test
from app.config import settings
settings.API_KEYS_RAW = VALID_API_KEY
with TestClient(app) as client:
yield client
app.dependency_overrides.clear()
def test_verify_api_key_required(client: TestClient):
# Sans clé API, l'accès doit être refusé (401)
response = client.get("/api/v1/tasks")
assert response.status_code == 401
assert "clé API est manquante" in response.json()["detail"]
def test_verify_api_key_invalid(client: TestClient):
# Avec une clé API incorrecte, l'accès doit être refusé (403)
response = client.get("/api/v1/tasks", headers={"X-API-Key": "mauvaise-cle"})
assert response.status_code == 403
assert "Clé API invalide" in response.json()["detail"]
def test_create_and_read_task(client: TestClient):
headers = {"X-API-Key": VALID_API_KEY}
# 1. Création d'une tâche
payload = {
"title": "Acheter du pain",
"description": "Prendre une tradition bien cuite",
"status": "A faire"
}
response = client.post("/api/v1/tasks", json=payload, headers=headers)
assert response.status_code == 201
data = response.json()
assert data["title"] == "Acheter du pain"
assert "id" in data
assert "created_at" in data
task_id = data["id"]
# 2. Lecture de la tâche spécifique
response = client.get(f"/api/v1/tasks/{task_id}", headers=headers)
assert response.status_code == 200
assert response.json()["title"] == "Acheter du pain"
def test_update_task_status(client: TestClient):
headers = {"X-API-Key": VALID_API_KEY}
# 1. Création
payload = {"title": "Coder l'API"}
response = client.post("/api/v1/tasks", json=payload, headers=headers)
task_id = response.json()["id"]
# 2. Passage à "En cours"
response = client.put(f"/api/v1/tasks/{task_id}", json={"status": "En cours"}, headers=headers)
assert response.status_code == 200
assert response.json()["status"] == "En cours"
# 3. Passage à "Termine"
response = client.put(f"/api/v1/tasks/{task_id}", json={"status": "Termine"}, headers=headers)
assert response.status_code == 200
assert response.json()["status"] == "Termine"
def test_delete_task(client: TestClient):
headers = {"X-API-Key": VALID_API_KEY}
# 1. Création
payload = {"title": "Tâche éphémère"}
response = client.post("/api/v1/tasks", json=payload, headers=headers)
task_id = response.json()["id"]
# 2. Suppression
response = client.delete(f"/api/v1/tasks/{task_id}", headers=headers)
assert response.status_code == 204
# 3. Vérification de la disparition (404)
response = client.get(f"/api/v1/tasks/{task_id}", headers=headers)
assert response.status_code == 404