johanleroy/ENI-PythonAdvanced_10
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

First commit

Browse Source
This commit is contained in:
Johan
2025-12-17 13:29:14 +01:00
commit cc50161771
43 changed files with 3665 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.env Normal file
View File

@@ -0,0 +1 @@
BACKEND_CORS_ORIGINS=["http://localhost:4200"]

1
.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
/.idea

296
README.md Normal file
View File

@@ -0,0 +1,296 @@
# TP : API IA d'analyse de films (GraphQL & LLM)
## Informations générales
**Cours** : Python Avancé > REST API > Fast API \
**Objectifs pédagogiques** :
- Mettre en place une API GraphQL avec Strawberry sur une base FastAPI existante.
- Structurer une application en couches (API, BLL/Services, DAL/Repository).
- Définir des schémas de données avec Pydantic pour la validation et la sérialisation.
- Interagir avec une API externe de manière asynchrone.
- Gérer les erreurs métier et les remonter proprement dans GraphQL.
- Intégrer un Large Language Model (LLM) pour des fonctionnalités d'IA.
- Optimiser les appels au LLM en fonction des champs demandés par le client GraphQL.
-----
## Objectif
L'objectif de ce TP est de construire une API GraphQL performante en utilisant FastAPI et Strawberry. Cette API exposera un endpoint capable d'analyser des films. Pour cela, elle ira chercher les informations d'un film depuis une API REST externe, puis utilisera un Large Language Model (LLM) local (via LM Studio) pour générer des analyses intelligentes et dynamiques, comme des résumés, des tags pertinents ou le genre le plus approprié.
Nous nous concentrerons sur l'implémentation des couches de services et de repositories, ainsi que sur la définition des types et resolvers GraphQL, en nous appuyant sur une structure de projet déjà initialisée.
-----
## Prérequis
* Avoir une instance de **LM Studio** (ainsi que le serveur associé) qui tourne localement, avec le modèle `meta-llama-3.1-8b-instruct` prêt à l'emploi. Attention : si votre machine ne supporte pas un tel modèle (pas de GPU dédié), utilisez un modèle fourni dans l'archive `02_Models_Light.zip`.
* Avoir accès à une **API REST de films** (supposée tourner sur `http://127.0.0.1:8000`). Cette API est considérée comme un prérequis et n'est pas à développer dans ce TP.
* Après avoir récupéré le projet, installez les dépendances via Poetry :
```bash
poetry install
```
* A ce stade, si vos imports ne sont pas reconnus dans l'IDE PyCharm à l'ouverture d'un fichier (par exemple `app/main.py`), marquer le répertoire `src` comme `Sources Root` (Clic droit sur répertoire `src`, puis `Mark Directory as ... > Sources Root`) ou redémarrer l'IDE (File > Invalidate Caches... > Just restart à gauche).
* Pour démarrer le serveur GraphQL, utilisez la commande suivante à la racine du projet :
```bash
# En étant à la racine du projet
cd src
uvicorn app.main:app --host 0.0.0.0 --port 8002
```
* Ouvrez un client GraphQL comme Apollo Studio Sandbox :
[https://studio.apollographql.com/sandbox/explorer/](https://studio.apollographql.com/sandbox/explorer/)
* Renseignez dans l'outil l'URL de votre serveur local : `http://127.0.0.1:8002/graphql`
À noter que vous pouvez aussi utiliser l'interface incluse GraphiQL en vous rendant à l'adresse : http://127.0.0.1:8002/graphql
Attention : si vous êtes derrière un proxy, vous devez définir en haut de votre fichier Python `main.py` :
```python
import os
os.environ["NO_PROXY"] = "127.0.0.1,localhost"
```
-----
## Étape 1 : Prise en main du projet et configuration
Le projet qui vous est fourni contient déjà une structure de base pour vous permettre de vous concentrer sur l'essentiel.
1. **Explorez la structure du projet** : vous remarquerez les fichiers suivants, déjà créés et configurés pour vous :
* `app/main.py` : point d'entrée de l'application FastAPI, avec le routeur GraphQL déjà configuré.
* `app/core/config.py` : gestion de la configuration via Pydantic et les variables d'environnement.
* `app/core/exceptions.py` : définition des exceptions personnalisées pour la couche métier (BLL) et la couche d'accès aux données (DAL).
* `app/core/llm.py` : initialisation du client pour communiquer avec le LLM.
* `app/graphql/context.py` : fonction pour injecter l'instance du LLM dans le contexte de chaque requête GraphQL.
* `app/graphql/extensions.py` : extension Strawberry pour une gestion propre des erreurs métier (`BLLException`).
* `app/graphql/mutations.py` : une mutation `_noop` pour assurer un schéma valide.
2. **Complétez l'arborescence de fichiers** : créez les dossiers et fichiers vides qui manquent pour atteindre la structure cible :
```
.
├── app
│ ├── core
│ │ ├── config.py # Fourni
│ │ ├── exceptions.py # Fourni
│ │ └── llm.py # Fourni
│ ├── graphql
│ │ ├── inputs
│ │ │ └── movie_input.py # Fourni
│ │ ├── resolvers
│ │ │ ├── analyze_movie_v1.py # A créer (étape 3)
│ │ │ ├── analyze_movie_v2.py # A créer (étape 6)
│ │ │ └── helper.py # Fourni
│ │ ├── types
│ │ │ └── movie_analysis.py # Fourni
│ │ ├── context.py # Fourni
│ │ ├── extensions.py # Fourni
│ │ ├── mutations.py # Fourni
│ │ └── queries.py # A créer (étape 3)
│ ├── models
│ │ ├── genre.py # Fourni
│ │ ├── member.py # Fourni
│ │ ├── movie.py # Fourni
│ │ ├── opinion.py # Fourni
│ │ └── person.py # Fourni
│ ├── repositories
│ │ ├── _base_client.py # Fourni
│ │ ├── genre_repository.py # A créer (étape 2)
│ │ └── movie_repository.py # A créer (étape 2)
│ └── services
│ ├── movie_analyzer_v1.py # A créer (étape 3)
│ └── movie_analyzer_v2.py # A créer (étape 5)
├── .env # A créer (étape 1) OU utiliser variables d'env.
└── main.py # Fourni (dans app/)
```
3. Créez un fichier **`.env`** à la racine du projet pour configurer les URLs. Cela permet de modifier les configurations sans toucher au code.
```.env
BACKEND_CORS_ORIGINS=["http://localhost:4200"] # En prévision de l'accès depuis un front Angular.
```
**Astuce** : vous pouvez aussi définir ces variables d'environnement directement dans votre terminal avant de lancer l'application sous PyCharm. Faites clic droit "Settings" sur le terminal, puis modifier ces variables dans l'onglet "Environment Variables".
---
## Étape 2 : couche d'accès aux données (Repositories)
La couche "Repository" est responsable de toute interaction avec des sources de données externes, ici notre API REST de films.
Complétez les `TODO` des **repositories** pour les films et les genres.
- `app/repositories/movie_repository.py` : contiendra les méthodes pour récupérer un ou plusieurs films. La méthode `find_by_id` devra retourner `None` si le film n'est pas trouvé (status code 404).
- `app/repositories/genre_repository.py` : contiendra la méthode pour lister tous les genres disponibles.
A ce stade, vous devriez pouvoir tester votre travail avec la commande suivante : ` poetry run pytest tests/etape2_dal_repositories.py`.
---
## Étape 3 : mise en place de GraphQL (Version 1 - Statique)
Nous allons construire la structure de notre API GraphQL avec une première version qui renvoie des données en dur, pour valider la mise en place.
1. Observez le **type de sortie GraphQL** dans `app/graphql/types/movie_analysis.py`. Ce sera l'objet que notre API retournera.
2. Complétez un premier **service d'analyse statique** dans `app/services/movie_analyzer_v1.py`. Ce service prendra un `movie_id` et retournera un dictionnaire avec des données pré-remplies, simulant une analyse IA.
3. Complétez le **resolver GraphQL** dans `app/graphql/resolvers/analyze_movie_v1.py`. Un resolver est une fonction qui sait comment obtenir les données pour un champ. Celui-ci appellera votre service `movie_analyzer_v1`.
```mermaid
graph TD
Client[Client GraphQL] -- "Query analyzeMovie v1" --> Resolver[Resolver analyze_movie_v1];
Resolver --> Service[Service movie_analyzer_v1];
Service --> Data["Données en dur 'mock'"];
Data --> Service;
Service --> Resolver;
Resolver -- "Réponse GraphQL" --> Client;
```
4. Exposez le resolver dans le schéma Mettez à jour le fichier `app/graphql/queries.py` pour importer votre resolver `analyze_movie_by_id` (depuis `app.graphql.resolvers.analyze_movie_v1`) et l'ajouter au type Query.
```python
# app/graphql/queries.py
import strawberry
# Importer le resolver V1
from app.graphql.resolvers.analyze_movie_v1 import analyze_movie_by_id
from app.graphql.types.movie_analysis import MovieAnalysis
@strawberry.type
class Query:
"""
Point d'entrée pour toutes les requêtes GraphQL de type 'query'.
"""
# Remplacer le TODO par ceci :
analyzeMovie: MovieAnalysis = strawberry.field(
resolver=analyze_movie_by_id,
description="Analyse un film en utilisant l'IA (version 1 statique)."
)
```
A ce stade, vous devriez pouvoir tester votre travail avec la commande suivante : `poetry run pytest tests/etape3_graphql_v1.py`.
À ce stade, vous devriez pouvoir lancer l'application et tester la query suivante dans votre client GraphQL pour recevoir les données statiques :
```graphql
query AnalyzeMovieByID {
analyzeMovie(movieId: "1") {
id
aiSummary
aiOpinionSummary
aiBestGenre
aiTags
}
}
```
À ce stade, **assurez-vous svp** que dans `app\graphql\queries.py`, vous ayez `from app.graphql.resolvers.analyze_movie_v1 import analyze_movie_by_id` (et non `from app.graphql.resolvers.analyze_movie_v2 import analyze_movie_by_id`)
---
## Étape 4 : Comprendre l'Intégration du LLM
Le mécanisme d'intégration du LLM est déjà en place. Prenez un moment pour comprendre comment il fonctionne :
1. **`app/core/llm.py`** : ce fichier crée une instance unique de `ChatOpenAI` en utilisant les paramètres du fichier `.env` (ou variables d'environnement, ou les valeurs par défaut de votre fichier `config.py`). Cette instance sera partagée par toute l'application.
2. **`app/graphql/context.py`** : la fonction `get_context` est appelée pour chaque requête GraphQL et y injecte notre instance `llm`.
3. **`app/main.py`** : lors de la création du `GraphQLRouter`, nous passons notre fonction `get_context`. Ainsi, chaque resolver pourra accéder au LLM via `info.context["llm"]`.
---
## Étape 5 : service d'analyse avancé (Version 2 - Dynamique)
Nous allons créer le service qui interagit réellement avec le LLM pour générer des analyses.
Complétez le fichier **`app/services/movie_analyzer_v2.py`**.
A ce stade, vous devriez pouvoir tester votre travail avec la commande suivante : `poetry run pytest tests/etape5_service_v2.py`.
---
## Étape 6 : optimisation du Resolver GraphQL (Version 2)
Un client GraphQL peut demander seulement un sous-ensemble des champs disponibles.
Nous allons optimiser notre resolver pour n'appeler le LLM que pour les champs réellement demandés.
En effet, l'appel à un LLM est coûteux en temps et en ressources, on souhaite donc éviter les appels inutiles.
1. Observez la **fonction utilitaire** dans `app/graphql/resolvers/helper.py` nommée `is_field_requested(info: Info, field_name: str) -> bool`.
2. Complétez le **nouveau resolver** dans `app/graphql/resolvers/analyze_movie_v2.py`. Il doit :
* récupérer l'instance `llm` depuis le contexte (`info.context["llm"]`).
* utiliser votre helper `is_field_requested` pour déterminer quels arguments booléens passer au service `movie_analyzer_v2`.
* appeler le service et retourner le résultat.
3. Mettez à jour **`app/graphql/queries.py`** pour qu'il utilise ce nouveau resolver `analyze_movie_v2`.
À ce stade, **assurez-vous svp** que dans `app\graphql\queries.py`, vous ayez `from app.graphql.resolvers.analyze_movie_v2 import analyze_movie_by_id` (et non `from app.graphql.resolvers.analyze_movie_v1 import analyze_movie_by_id`)
---
```mermaid
flowchart TD
subgraph "Etape 1: Resolver"
Client[Client GraphQL] -- "1. Query { aiSummary, aiTags }" --> Resolver[Resolver: analyze_movie_by_id];
Resolver -- "2. Récupère flags 'info'" --> Flags["flags: ai_summary=True, ai_tags=True..."];
Flags -- "3. Appelle service" --> Service[Service: analyze_movie];
end
subgraph "Etape 2: Service 'analyze_movie'"
Service -- "4. 'ai_best_genre' == True?" --> CheckGenre{Flag: ai_best_genre?};
CheckGenre -- "5a. True" --> FetchGenres[Repo: genre_repository.list];
CheckGenre -- "5b. False" --> FetchMovie[Repo: movie_repository.find_by_id];
FetchGenres -- "6. Récupère film" --> FetchMovie;
FetchMovie -- "7. Données film + genres" --> BuildTasks[Construction dict 'tasks'];
BuildTasks -- "8a. if 'ai_summary' True" --> TaskSummary["Ajout get_ai_summary"];
BuildTasks -- "8b. if 'ai_opinion' True" --> TaskOpinion["Ajout get_ai_opinion_summary"];
BuildTasks -- "8c. if 'ai_best_genre' True" --> TaskGenre["Ajout get_ai_best_genre"];
BuildTasks -- "8d. if 'ai_tags' True" --> TaskTags["Ajout get_ai_tags"];
TaskSummary -- "Coroutines" --> Gather[Exécution: asyncio.gather];
TaskOpinion -- "Coroutines" --> Gather;
TaskGenre -- "Coroutines" --> Gather;
TaskTags -- "Coroutines" --> Gather;
BuildTasks -- "8e. si 'tasks' vide" --> Combine[Combiner résultats];
end
subgraph "Etape 3: Appels LLM Parallèles"
Gather -- "9. Lance tâches" --> LLM_Calls[Appels LLM concurrents];
LLM_Calls --> LLM1["LLM: get_ai_summary"];
LLM_Calls --> LLM_Opinion["LLM: get_ai_opinion_summary"];
LLM_Calls --> LLM_Genre["LLM: get_ai_best_genre"];
LLM_Calls --> LLM_Tags["LLM: get_ai_tags"];
end
LLM1 -- "10. Résultat" --> Combine;
LLM_Opinion -- "10. Résultat" --> Combine;
LLM_Genre -- "10. Résultat" --> Combine;
LLM_Tags -- "10. Résultat" --> Combine;
FetchMovie -- "Données film (ID)" --> Combine;
Combine -- "11. Dict final" --> Service;
Service -- "12. Réponse" --> Resolver;
Resolver -- "13. Objet MovieAnalysis" --> Client;
```
A ce stade, vous devriez pouvoir tester votre travail avec la commande suivante : `poetry run pytest tests/etape6_resolver_v2.py`.
-----
## Étape 7 (bonus) : Comprendre la gestion fine des erreurs métier
Le mécanisme pour retourner des erreurs claires au client est déjà en place.
1. **Examinez `app/graphql/extensions.py`** : la classe `BusinessLogicErrorExtension` intercepte les erreurs. Si une erreur est une instance de `BLLException` (comme `NotFoundBLLException`), elle reformate le message et ajoute un code d'erreur dans la réponse GraphQL.
2. **Examinez `app/main.py`** : vous verrez que cette extension est passée à `strawberry.Schema` lors de sa création, ce qui l'active pour toutes les requêtes.
Pour tester, essayez de demander un film avec un ID qui n'existe pas. Vous devriez voir une erreur GraphQL formatée.
A ce stade, vous devriez pouvoir tester votre travail avec la commande suivante : `poetry run pytest tests/etape7_erreurs.py`.
---
## Conclusion
Vous avez construit une API GraphQL complète, asynchrone et optimisée qui s'appuie sur la puissance d'un LLM pour fournir des données enrichies. Vous avez mis en pratique des concepts d'architecture logicielle, de programmation asynchrone et d'optimisation des requêtes.

2194
poetry.lock generated Normal file
View File

File diff suppressed because it is too large Load Diff

65
pyproject.toml Normal file
View File

@@ -0,0 +1,65 @@
[project]
name = "tp-fastapi-graphql"
version = "0.1.0"
description = "TP FastAPI GraphQL"
authors = [
{name = "Your Name",email = "you@example.com"}
]
readme = "README.md"
classifiers = [
"Programming Language :: Python :: 3.13",
]
keywords = ["fastapi", "graphql", "web", "ia", "llm"]
exclude = [
{ path = "tests", format = "wheel" }
]
requires-python = ">=3.13"
[tool.poetry]
package-mode = false
[tool.poetry.dependencies]
python = "^3.13"
fastapi = "^0.116.1"
uvicorn = { version = "^0.35.0", extras = [ "standard" ] }
gunicorn = "^23.0.0"
pydantic = "^2.11.7"
python-dotenv = "^1.0.1"
pydantic-settings = "^2.10.1"
httpx = "^0.28.1"
langchain = "^0.3.27"
langchain-openai = "^0.3.35"
strawberry-graphql = "^0.281.0"
[tool.poetry.group.dev.dependencies]
pytest = "^8.4.1"
pytest-cov = "^6.2.1"
pytest-asyncio = "^1.1.0"
pytest-mock = "^3.14.1"
aiosqlite = "^0.21.0"
coverage = { version="*", extras=["toml"]}
[tool.pytest.ini_options]
asyncio_mode = "auto"
pythonpath = "src"
testpaths = "tests"
addopts = "-v -s"
[tool.black]
line-length = 120
[tool.pycln]
all = true
[tool.isort]
line_length = 120
multi_line_output = 3
include_trailing_comma = true
force_grid_wrap = 0
use_parentheses = true
ensure_newline_before_comments = true
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"

0
src/app/__init__.py Normal file
View File

0
src/app/core/__init__.py Normal file
View File

34
src/app/core/config.py Normal file
View File

@@ -0,0 +1,34 @@
from typing import List
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""
Classe de configuration qui charge les variables d'environnement.
"""
# Configuration du modèle Pydantic
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=True
)
# Paramètres du projet
PROJECT_NAME: str = "FastAPI GraphQL LLM Project"
# Configuration de LM Studio (serveur Chat local)
LLM_CHAT_SERVER_BASE_URL: str = "http://127.0.0.1:1234/v1"
LLM_CHAT_MODEL: str = "meta-llama-3.1-8b-instruct"
LLM_CHAT_TEMPERATURE: float = 0.3 # On baisse un peu la température pour des résultats plus prévisibles (mais moins créatifs)
LLM_CHAT_API_KEY: str = "not-needed" # Clé API factice, LM Studio ne l'utilise pas
# Configuration de l'API
MOVIE_API_BASE_URL: str = "http://127.0.0.1:8000/api/v1"
# Configuration CORS
# Pydantic va automatiquement convertir la chaîne de caractères séparée par des virgules
# en une liste de chaînes de caractères.
BACKEND_CORS_ORIGINS: List[AnyHttpUrl] = []
settings = Settings()

25
src/app/core/exceptions.py Normal file
View File

@@ -0,0 +1,25 @@
class BaseAppException(Exception):
"""Exception de base pour l'application."""
pass
class DALException(BaseAppException):
"""Exception levée pour les erreurs de la couche d'accès aux données (DAL)."""
def __init__(self, message: str, original_exception: Exception = None):
self.message = message
self.original_exception = original_exception
super().__init__(self.message)
class BLLException(BaseAppException):
"""Exception de base pour les erreurs de la couche métier (BLL)."""
pass
class NotFoundBLLException(BLLException):
"""Levée lorsqu'une ressource n'est pas trouvée."""
def __init__(self, resource_name: str, resource_id: int | str):
message = f"{resource_name} avec l'ID '{resource_id}' non trouvé."
super().__init__(message)
class ValidationBLLException(BLLException):
"""Levée pour les erreurs de validation des règles métier."""
pass

11
src/app/core/llm.py Normal file
View File

@@ -0,0 +1,11 @@
from app.core.config import settings
from langchain_openai import ChatOpenAI
# --- Initialisation du Modèle de Langage (LLM) ---
# Cette instance unique sera créée au démarrage de l'application et partagée par toutes les requêtes.
llm = ChatOpenAI(
model=settings.LLM_CHAT_MODEL,
base_url=settings.LLM_CHAT_SERVER_BASE_URL,
temperature=settings.LLM_CHAT_TEMPERATURE,
api_key=settings.LLM_CHAT_API_KEY
)

0
src/app/graphql/__init__.py Normal file
View File

12
src/app/graphql/context.py Normal file
View File

@@ -0,0 +1,12 @@
from fastapi import Request
from typing import Any, Dict
from app.core.llm import llm
async def get_context(request: Request) -> Dict[str, Any]:
"""
Crée le contexte pour chaque requête GraphQL.
"""
return {
"request": request,
"llm": llm # on ajoute l'instance llm au dictionnaire du contexte
}

15
src/app/graphql/extensions.py Normal file
View File

@@ -0,0 +1,15 @@
from strawberry.extensions import Extension
from app.core.exceptions import BLLException
class BusinessLogicErrorExtension(Extension):
def on_request_end(self):
# On parcourt les erreurs qui ont pu se produire
for error in self.execution_context.errors:
original_error = error.original_error
# Si l'erreur est une de nos exceptions métier...
if isinstance(original_error, BLLException):
# ...on peut reformater le message et ajouter des extensions
error.message = f"[Business Error] {original_error}"
if not error.extensions:
error.extensions = {}
error.extensions['code'] = original_error.__class__.__name__

0
src/app/graphql/inputs/__init__.py Normal file
View File

11
src/app/graphql/inputs/movie_input.py Normal file
View File

@@ -0,0 +1,11 @@
import strawberry
from typing import Optional
# Cette classe utilisée par les resolvers, représente l'input pour une opération liée à un film
# Elle contient uniquement l'ID du film
# Elle est donnée à titre indicatif et peut être étendue avec d'autres champs si nécessaire.
@strawberry.input
class MovieInput:
id: strawberry.ID

14
src/app/graphql/mutations.py Normal file
View File

@@ -0,0 +1,14 @@
import strawberry
from app.graphql.resolvers.noop import resolve_noop
@strawberry.type
class Mutation:
"""
Point d'entrée pour toutes les mutations GraphQL.
"""
_noop: bool = strawberry.field(
resolver=resolve_noop,
description="Mutation factice pour assurer la validité du schéma."
)

23
src/app/graphql/queries.py Normal file
View File

@@ -0,0 +1,23 @@
import strawberry
# Note : l'énoncé demande d'importer la V1 à l'étape 3, puis la V2 à l'étape 6
# L'import ci-dessous correspond à l'étape 6.
from app.graphql.resolvers.analyze_movie_v2 import analyze_movie_by_id
from app.graphql.types.movie_analysis import MovieAnalysis
@strawberry.type
class Query:
"""
Point d'entrée pour toutes les requêtes GraphQL de type 'query'.
"""
# TODO : (Étape 3) L'énoncé vous demande d'implémenter ce champ
# en important et en utilisant 'analyze_movie_v1'
# (Étape 6) L'énoncé vous demande ensuite de basculer vers 'analyze_movie_v2'
# La configuration ci-dessous correspond à l'étape 6.
analyzeMovie: MovieAnalysis = strawberry.field(
resolver=analyze_movie_by_id, # 'analyze_movie_by_id' est importé depuis v2
description="Analyse un film en utilisant l'IA."
)

0
src/app/graphql/resolvers/__init__.py Normal file
View File

18
src/app/graphql/resolvers/analyze_movie_v1.py Normal file
View File

@@ -0,0 +1,18 @@
import strawberry
from strawberry import Info
# from app.graphql.inputs.movie_input import MovieInput
from app.graphql.types.movie_analysis import MovieAnalysis
from app.services.movie_analyzer_v1 import analyze_movie
async def analyze_movie_by_id(
movie_id: strawberry.ID,
# movie_input: MovieInput, classe nécessaire seulement si on avait eu beaucoup de champs en entrée
) -> MovieAnalysis:
# TODO : (Étape 3) Remplacer la ligne suivante par un appel au service V1
# analysis_data = await analyze_movie(movie_id=movie_id)
raise NotImplementedError("Le resolver analyze_movie_v1.analyze_movie_by_id() n'est pas implémenté.")
# La ligne ci-dessous doit être décommentée une fois le TODO complété
# return MovieAnalysis(**analysis_data)

29
src/app/graphql/resolvers/analyze_movie_v2.py Normal file
View File

@@ -0,0 +1,29 @@
import strawberry
from strawberry import Info
from app.graphql.resolvers.helper import is_field_requested
from app.graphql.types.movie_analysis import MovieAnalysis
from app.services.movie_analyzer_v2 import analyze_movie
async def analyze_movie_by_id(
movie_id: strawberry.ID,
info: Info,
) -> MovieAnalysis:
# TODO : (Étape 6) Remplacer la ligne suivante par la récupération
# du LLM depuis 'info.context' (ex: llm = info.context["llm"])
raise NotImplementedError("Le resolver V2 n'a pas encore récupéré le LLM du contexte.")
# llm = ... # <== Code à écrire
# Le 'llm=llm' ci-dessous fonctionnera une fois le TODO complété
analysis_data = await analyze_movie(
movie_id=movie_id,
ai_summary=is_field_requested(info, "aiSummary"),
ai_opinion_summary=is_field_requested(info, "aiOpinionSummary"),
ai_best_genre=is_field_requested(info, "aiBestGenre"),
ai_tags=is_field_requested(info, "aiTags"),
llm=llm
)
return MovieAnalysis(**analysis_data)

11
src/app/graphql/resolvers/helper.py Normal file
View File

@@ -0,0 +1,11 @@
from strawberry import Info
def is_field_requested(info: Info, field_name: str) -> bool:
for field_node in info.field_nodes:
if field_node.selection_set:
for selection in field_node.selection_set.selections:
if selection.name.value == field_name:
return True
return False

6
src/app/graphql/resolvers/noop.py Normal file
View File

@@ -0,0 +1,6 @@
def resolve_noop() -> bool:
"""
Resolver pour la mutation _noop.
Ne fait rien d'autre que de retourner True.
"""
return True

0
src/app/graphql/types/__init__.py Normal file
View File

12
src/app/graphql/types/movie_analysis.py Normal file
View File

@@ -0,0 +1,12 @@
from typing import List, Optional
import strawberry
@strawberry.type
class MovieAnalysis:
id: strawberry.ID
aiSummary: Optional[str]
aiOpinionSummary: Optional[str]
aiBestGenre: Optional[str]
aiTags: Optional[List[str]]

44
src/app/main.py Normal file
View File

@@ -0,0 +1,44 @@
from fastapi import FastAPI
from app.core.config import settings
from strawberry.fastapi import GraphQLRouter
import strawberry
import uvicorn
from fastapi.middleware.cors import CORSMiddleware
from app.graphql.context import get_context
from app.graphql.extensions import BusinessLogicErrorExtension
from app.graphql.mutations import Mutation
from app.graphql.queries import Query
# Crée l'application FastAPI
app = FastAPI(
title="Movie AI GraphQL API",
description="API GraphQL pour l'analyse de films par IA",
version="1.0.0"
)
# CORS middleware pour permettre les requêtes depuis n'importe quelle origine
app.add_middleware(
CORSMiddleware,
allow_origins=[str(origin).rstrip('/') for origin in settings.BACKEND_CORS_ORIGINS] if settings.BACKEND_CORS_ORIGINS else ["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Crée le schéma GraphQL avec les types de requêtes et de mutations
schema = strawberry.Schema(
query=Query,
mutation=Mutation,
extensions=[BusinessLogicErrorExtension]
)
# Crée le routeur GraphQL et l'ajoute à l'application
graphql_app = GraphQLRouter(schema, context_getter=get_context)
app.include_router(graphql_app, prefix="/graphql")
@app.get("/", tags=["Root"])
def read_root():
return {"message": "Bienvenue sur l'API de l'Analyseur IA de films. Rendez-vous sur /graphql"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8002)

0
src/app/models/__init__.py Normal file
View File

5
src/app/models/genre.py Normal file
View File

@@ -0,0 +1,5 @@
from pydantic import BaseModel, ConfigDict
class Genre(BaseModel):
id: int
label: str

5
src/app/models/member.py Normal file
View File

@@ -0,0 +1,5 @@
from pydantic import BaseModel
class Member(BaseModel):
id: int
login: str

19
src/app/models/movie.py Normal file
View File

@@ -0,0 +1,19 @@
from pydantic import BaseModel
from typing import List, Optional
from app.models.genre import Genre
from app.models.opinion import Opinion
from app.models.person import Person
class Movie(BaseModel):
id: int
title: str
year: int
duration: Optional[int] = None
synopsis: Optional[str] = None
genre: Genre
director: Person
actors: List[Person]
opinions: List[Opinion]

11
src/app/models/opinion.py Normal file
View File

@@ -0,0 +1,11 @@
from pydantic import BaseModel
from app.models.member import Member
class Opinion(BaseModel):
id: int
note: int
comment: str
movie_id: int
member: Member

7
src/app/models/person.py Normal file
View File

@@ -0,0 +1,7 @@
from pydantic import BaseModel
from typing import Optional
class Person(BaseModel):
id: int
last_name: str
first_name: Optional[str] = None

0
src/app/repositories/__init__.py Normal file
View File

33
src/app/repositories/_base_client.py Normal file
View File

@@ -0,0 +1,33 @@
import httpx
from app.core.config import settings
from app.core.exceptions import DALException
class BaseClient:
"""Client de base pour gérer les appels HTTP et les erreurs."""
def __init__(self, base_url: str):
self.client = httpx.AsyncClient(base_url=base_url)
async def _request(self, method: str, url: str, **kwargs) -> httpx.Response:
"""Effectue une requête et gère les exceptions communes."""
try:
response = await self.client.request(method, url, **kwargs)
response.raise_for_status()
return response
except httpx.HTTPStatusError as e:
# Erreur HTTP (4xx, 5xx)
dal_exception = DALException(
message=f"Erreur API: {e.response.status_code} pour {e.request.url}. Réponse: {e.response.text}",
original_exception=e
)
dal_exception.status_code = e.response.status_code
raise dal_exception from e
except httpx.RequestError as e:
# Erreur réseau (timeout, connexion impossible...)
raise DALException(
message=f"Erreur réseau pour {e.request.url}",
original_exception=e
) from e
api_client = BaseClient(base_url=settings.MOVIE_API_BASE_URL)

16
src/app/repositories/genre_repository.py Normal file
View File

@@ -0,0 +1,16 @@
from typing import List
from app.models.genre import Genre
from app.repositories._base_client import api_client
class GenreRepository:
async def list(self) -> List[Genre]:
response = await api_client._request("GET", "/genres/")
# TODO : (Étape 2) Remplacer la ligne suivante par un appel à
# response = await api_client._request("GET", "/genres/")
# La ligne ci-dessous doit être décommentée une fois le TODO complété
return [Genre.model_validate(g) for g in response.json()]
genre_repository = GenreRepository()

28
src/app/repositories/movie_repository.py Normal file
View File

@@ -0,0 +1,28 @@
from typing import List, Optional
import httpx
from app.core.exceptions import DALException
from app.models.movie import Movie
from app.repositories._base_client import api_client
class MovieRepository:
async def list(self, skip: int = 0, limit: int = 100) -> List[Movie]:
# TODO : (Étape 2) Remplacer la ligne suivante par un appel à
# response = await api_client._request("GET", "/movies/", params={"skip": skip, "limit": limit})
# La ligne ci-dessous doit être décommentée une fois le TODO complété
response = await api_client._request("GET", "/movies/", params={"skip": skip, "limit": limit})
return [Movie.model_validate(m) for m in response.json()]
async def find_by_id(self, movie_id: int) -> Optional[Movie]:
try:
response = await api_client._request("GET", f"/movies/{movie_id}")
# TODO : (Étape 2) Remplacer la ligne suivante par un appel à
# response = await api_client._request("GET", f"/movies/{movie_id}")
# La ligne ci-dessous doit être décommentée une fois le TODO complété
return Movie.model_validate(response.json())
except DALException as e:
if e.status_code == 404:
return None
raise
movie_repository = MovieRepository()

0
src/app/services/__init__.py Normal file
View File

17
src/app/services/movie_analyzer_v1.py Normal file
View File

@@ -0,0 +1,17 @@
async def analyze_movie(movie_id: str) -> dict:
# TODO : retourner un dictionnaire python statique (chaînes de caractères en dur) avec comme attributs:
# id (correspondant à movie_id)
# aiSummary (chaîne de caractères arbitraire)
# aiOpinionSummary (chaîne de caractères arbitraire)
# aiBestGenre (chaîne de caractères arbitraire)
# aiTags (TABLEAU de chaînes de caractères arbitraire)
raise NotImplementedError("Le service movie_analyzer_v1.analyze_movie() n'est pas implémenté.")
return {
"id": movie_id,
}

143
src/app/services/movie_analyzer_v2.py Normal file
View File

@@ -0,0 +1,143 @@
import asyncio
import strawberry
from langchain_core.language_models import BaseChatModel
from app.core.exceptions import NotFoundBLLException
from app.repositories.movie_repository import movie_repository
from app.repositories.genre_repository import genre_repository
async def get_ai_summary(llm, synopsis):
if not synopsis:
return None
prompt = f"""
Français uniquement.
Fais un résumé très court (une à deux phrases maximum) du synopsis suivant.
Ne retourne que le résumé, sans aucune phrase d'introduction comme "Voici le résumé :".
Synopsis : {synopsis}
"""
response = await llm.ainvoke(prompt)
return response.content.strip()
async def get_ai_opinion_summary(llm, title, opinions):
if not opinions:
return None
opinions_text = "\n".join([f"ID Opinion = {opinion.id}; Note : {opinion.note}/5; Commentaire : {opinion.comment}" for opinion in opinions])
prompt = f"""
Français uniquement.
Fais un résumé très court (une à deux phrases maximum) des opinions suivantes. Les opinions portent sur un même et unique film, dont le titre est : {title}.
Ne fais pas une liste d'items. Ne fais pas un résumé de chaque opinion individuellement, mais un résumé global.
Ne retourne que le résumé, sans aucune phrase d'introduction comme "Voici le résumé :".
Opinions :
{opinions_text}
"""
response = await llm.ainvoke(prompt)
return response.content.strip()
async def get_ai_best_genre(llm, synopsis, all_genres):
if not synopsis or not all_genres:
return None
# Préparation de la liste des genres pour le prompt
genres_list = ", ".join([genre.label for genre in all_genres])
# Prompt pour choisir le genre le plus pertinent
# TODO : compléter les instructions du prompt
prompt = f"""
# TODO : Écrire les instructions pour le LLM.
# Objectif : Choisir le *seul* genre le plus pertinent pour le film.
# Contraintes :
# 1. Le LLM DOIT répondre en français.
# 2. Le LLM DOIT choisir son genre EXCLUSIVEMENT parmi la liste fournie.
# 3. Le LLM NE DOIT retourner QUE le nom du genre (ex: "Drame"), sans aucune autre phrase.
Voici le synopsis :
{synopsis}
Voici la liste des genres autorisés :
{genres_list}
Genre le plus pertinent :
"""
# Appel asynchrone au modèle de langage
response = await llm.ainvoke(prompt)
return response.content.strip()
async def get_ai_tags(llm, title, synopsis):
if not title or not synopsis:
return None
# TODO : définir le prompt approprié
prompt = f"""
# TODO : Écrire les instructions pour le LLM.
# Objectif : Générer 5 tags (mots-clés) pertinents pour le film.
# Contraintes :
# 1. Le LLM DOIT répondre en français.
# 2. Le LLM DOIT retourner une liste de tags séparés par des virgules.
# 3. Le LLM NE DOIT PAS inclure de phrase d'introduction (ex: "Voici les tags :").
Titre du film : {title}
Synopsis : {synopsis}
Génère 5 tags pertinents, séparés par des virgules :
"""
response = await llm.ainvoke(prompt)
tags = [tag.strip() for tag in response.content.split(',') if tag.strip()]
return tags
async def analyze_movie(
movie_id : str,
ai_summary: bool,
ai_opinion_summary : bool,
ai_best_genre : bool,
ai_tags : bool,
llm: BaseChatModel
) -> dict:
# Récupération des genres
all_genres = []
if ai_best_genre:
all_genres = await genre_repository.list()
# Récupération des données du film
movie_data = await movie_repository.find_by_id(movie_id)
if not movie_data:
raise NotFoundBLLException(resource_name="Movie", resource_id=movie_id)
# Tâches à effectuer
tasks = {}
if ai_summary:
tasks["aiSummary"] = get_ai_summary(llm, movie_data.synopsis)
# TODO : (Étape 5) compléter la logique d'ajout des tâches avec :
# appeler 'get_ai_opinion_summary', 'get_ai_best_genre', 'get_ai_tags'
# mettre le résultat respectivement dans la clé "aiOpinionSummary", "aiBestGenre", "aiTags" (ATTENTION : il faut respecter la casse pour ces clés!) du tableau associatif (dictionnaire) "tasks"
# respectivement en fonction des booléens 'ai_opinion_summary', 'ai_best_genre', 'ai_tags'
if ai_opinion_summary or ai_best_genre or ai_tags:
raise NotImplementedError("La logique d'ajout de tâches (opinion, genre, tags) n'est pas implémentée.")
if tasks:
# On récupère les coroutines (les fonctions async prêtes à être lancées)
coroutines = tasks.values()
# On les exécute toutes en parallèle et on attend les résultats
results = await asyncio.gather(*coroutines)
# On associe les résultats aux clés que nous avons définies
result_map = dict(zip(tasks.keys(), results))
else:
result_map = {} # Aucune tâche n'a été demandée
output = {
'id': strawberry.ID(movie_id),
'aiSummary': result_map.get("aiSummary"),
'aiOpinionSummary': result_map.get("aiOpinionSummary"),
'aiBestGenre': result_map.get("aiBestGenre"),
'aiTags': result_map.get("aiTags")
}
print(output)
return output

140
tests/etape2_dal_repositories.py Normal file
View File

@@ -0,0 +1,140 @@
"""
Tests de validation pour l'Étape 2 : Couche d'accès aux données (Repositories)
[Version Corrigée]
Cette version corrige les assertions pour accepter les arguments positionnels (args)
au lieu de forcer les arguments mots-clés (kwargs) lors de l'appel à _request.
Elle corrige également la simulation de DALException pour qu'elle corresponde
à la logique de _base_client.py.
"""
import pytest
import pytest_asyncio
from unittest.mock import MagicMock, AsyncMock
from app.core.exceptions import DALException
from app.models.movie import Movie
from app.models.genre import Genre
# --- Fixtures de données (simule les réponses de l'API) ---
@pytest.fixture
def mock_movie_data():
"""Données JSON brutes pour un film, simulées depuis l'API."""
return {
"id": 1,
"title": "Inception",
"year": 2010,
"duration": 148,
"synopsis": "Un voleur...",
"genre": {"id": 1, "label": "Science-Fiction"},
"director": {"id": 1, "last_name": "Nolan", "first_name": "Christopher"},
"actors": [{"id": 2, "last_name": "DiCaprio", "first_name": "Leonardo"}],
"opinions": []
}
@pytest.fixture
def mock_genre_list_data():
"""Données JSON brutes pour une liste de genres."""
return [
{"id": 1, "label": "Science-Fiction"},
{"id": 2, "label": "Drame"}
]
# --- Tests ---
@pytest.mark.asyncio
async def test_movie_repo_find_by_id_success(mocker, mock_movie_data):
"""
Vérifie que movie_repository.find_by_id retourne un objet Movie
en cas de succès.
"""
# 1. Mock de l'api_client
mock_response = MagicMock()
mock_response.json.return_value = mock_movie_data
mock_api_client = AsyncMock()
mock_api_client._request.return_value = mock_response
# On "patch" l'instance importée dans le module du repository
mocker.patch('app.repositories.movie_repository.api_client', mock_api_client)
# 2. Appel de la méthode à tester
from app.repositories.movie_repository import movie_repository
movie = await movie_repository.find_by_id(1)
# 3. Assertions
mock_api_client._request.assert_called_once_with("GET", "/movies/1")
assert movie is not None
assert isinstance(movie, Movie)
assert movie.title == "Inception"
@pytest.mark.asyncio
async def test_movie_repo_find_by_id_not_found(mocker):
"""
Vérifie que movie_repository.find_by_id retourne None si l'API
lève une DALException avec un status 404.
"""
# 1. Mock de l'api_client pour qu'il lève une erreur 404
mock_api_client = AsyncMock()
# 1. Crée l'exception
mock_exception = DALException("Not Found")
# 2. Attache le status_code
mock_exception.status_code = 404
# 3. La définit comme side_effect
mock_api_client._request.side_effect = mock_exception
mocker.patch('app.repositories.movie_repository.api_client', mock_api_client)
# 2. Appel de la méthode à tester
from app.repositories.movie_repository import movie_repository
movie = await movie_repository.find_by_id(999)
# 3. Assertions
mock_api_client._request.assert_called_once_with("GET", "/movies/999")
assert movie is None
@pytest.mark.asyncio
async def test_movie_repo_list(mocker, mock_movie_data):
"""
Vérifie que movie_repository.list retourne une liste de Movies
et passe correctement les paramètres skip/limit.
"""
# 1. Mock de l'api_client
mock_response = MagicMock()
mock_response.json.return_value = [mock_movie_data] # L'API retourne une liste
mock_api_client = AsyncMock()
mock_api_client._request.return_value = mock_response
mocker.patch('app.repositories.movie_repository.api_client', mock_api_client)
# 2. Appel de la méthode à tester
from app.repositories.movie_repository import movie_repository
movies = await movie_repository.list(skip=5, limit=10)
# 3. Assertions
expected_params = {"skip": 5, "limit": 10}
mock_api_client._request.assert_called_once_with("GET", "/movies/", params=expected_params)
assert isinstance(movies, list)
assert isinstance(movies[0], Movie)
assert movies[0].id == 1
@pytest.mark.asyncio
async def test_genre_repo_list(mocker, mock_genre_list_data):
"""
Vérifie que genre_repository.list retourne une liste de Genres.
"""
# 1. Mock de l'api_client
mock_response = MagicMock()
mock_response.json.return_value = mock_genre_list_data
mock_api_client = AsyncMock()
mock_api_client._request.return_value = mock_response
mocker.patch('app.repositories.genre_repository.api_client', mock_api_client)
# 2. Appel de la méthode à tester
from app.repositories.genre_repository import genre_repository
genres = await genre_repository.list()
# 3. Assertions
mock_api_client._request.assert_called_once_with("GET", "/genres/")
assert isinstance(genres, list)
assert isinstance(genres[0], Genre)
assert genres[0].label == "Science-Fiction"

70
tests/etape3_graphql_v1.py Normal file
View File

@@ -0,0 +1,70 @@
"""
Tests de validation pour l'Étape 3 : Mise en place GraphQL (Version 1 - Statique)
Objectif :
1. Vérifier que le service `movie_analyzer_v1` retourne bien un dict "mock".
2. Vérifier que le resolver `analyze_movie_v1` appelle ce service et
construit correctement l'objet `MovieAnalysis`.
Prérequis :
- Les TODOs de `app/services/movie_analyzer_v1.py` sont complétés.
- Les TODOs de `app/graphql/resolvers/analyze_movie_v1.py` sont complétés.
- Le champ `analyzeMovie` est bien ajouté à `app/graphql/queries.py`
(en utilisant le resolver V1).
"""
import pytest
import strawberry
from unittest.mock import AsyncMock
from app.graphql.types.movie_analysis import MovieAnalysis
@pytest.mark.asyncio
async def test_service_v1_analyze_movie():
"""
Teste le service V1. Il doit retourner un dictionnaire
contenant les clés requises et l'ID correct.
"""
from app.services.movie_analyzer_v1 import analyze_movie
movie_id = "123"
analysis_data = await analyze_movie(movie_id)
assert isinstance(analysis_data, dict)
assert analysis_data["id"] == movie_id
assert "aiSummary" in analysis_data
assert "aiOpinionSummary" in analysis_data
assert "aiBestGenre" in analysis_data
assert "aiTags" in analysis_data
assert isinstance(analysis_data["aiTags"], list)
@pytest.mark.asyncio
async def test_resolver_v1_analyze_movie_by_id(mocker):
"""
Teste le resolver V1. Il doit appeler le service V1 et
retourner un objet `MovieAnalysis` typé.
"""
# 1. Mock du service V1 que le resolver est censé appeler
mock_service_data = {
"id": "1",
"aiSummary": "Mock summary",
"aiOpinionSummary": "Mock opinion",
"aiBestGenre": "Mock genre",
"aiTags": ["mock", "test"]
}
mock_analyze_service = AsyncMock(return_value=mock_service_data)
mocker.patch('app.graphql.resolvers.analyze_movie_v1.analyze_movie', mock_analyze_service)
# 2. Appel du resolver
from app.graphql.resolvers.analyze_movie_v1 import analyze_movie_by_id
result = await analyze_movie_by_id(movie_id=strawberry.ID("1"))
# 3. Assertions
# Vérifie que le service a bien été appelé
mock_analyze_service.assert_called_once_with(movie_id="1")
# Vérifie que le résultat est du bon type
assert isinstance(result, MovieAnalysis)
assert result.id == strawberry.ID("1")
assert result.aiSummary == "Mock summary"
assert result.aiTags == ["mock", "test"]

188
tests/etape5_service_v2.py Normal file
View File

@@ -0,0 +1,188 @@
"""
Tests de validation pour l'Étape 5 : Service d'analyse avancé (Version 2 - Dynamique)
Objectif :
1. Vérifier que les fonctions `get_ai_...` appellent bien le LLM.
2. Vérifier que la fonction principale `analyze_movie` (V2)
n'appelle QUE les fonctions nécessaires (en fonction des booléens).
3. Vérifier que les appels aux repositories sont corrects.
Prérequis :
- Les TODOs de `app/services/movie_analyzer_v2.py` sont complétés.
"""
import pytest
import strawberry
from unittest.mock import AsyncMock, MagicMock, patch
from app.models.movie import Movie
from app.models.genre import Genre
from app.models.person import Person
from app.models.opinion import Opinion
from app.models.member import Member
# --- Fixtures ---
@pytest.fixture
def mock_llm():
"""Fixture pour un LLM mocké."""
llm = MagicMock()
# Simule la réponse de llm.ainvoke(...)
mock_response = MagicMock()
mock_response.content = "Réponse du LLM"
llm.ainvoke = AsyncMock(return_value=mock_response)
return llm
@pytest.fixture
def mock_movie():
"""Fixture pour un objet Movie Pydantic complet."""
return Movie(
id=1,
title="Inception",
year=2010,
synopsis="Un voleur qui vole des secrets...",
genre=Genre(id=1, label="Science-Fiction"),
director=Person(id=1, last_name="Nolan", first_name="Christopher"),
actors=[Person(id=2, last_name="DiCaprio", first_name="Leonardo")],
opinions=[
Opinion(id=1, note=5, comment="Génial!", movie_id=1, member=Member(id=1, login="user1"))
]
)
@pytest.fixture
def mock_genres():
"""Fixture pour une liste d'objets Genre Pydantic."""
return [
Genre(id=1, label="Science-Fiction"),
Genre(id=2, label="Drame")
]
# --- Tests des helpers LLM ---
@pytest.mark.asyncio
async def test_get_ai_summary(mock_llm):
"""Teste le prompt de résumé."""
from app.services.movie_analyzer_v2 import get_ai_summary
synopsis = "Un long synopsis..."
result = await get_ai_summary(mock_llm, synopsis)
assert result == "Réponse du LLM"
mock_llm.ainvoke.assert_called_once()
prompt_call = mock_llm.ainvoke.call_args[0][0]
assert synopsis in prompt_call # Vérifie que le synopsis est dans le prompt
@pytest.mark.asyncio
async def test_get_ai_summary_no_synopsis(mock_llm):
"""Teste que le LLM n'est pas appelé si le synopsis est vide."""
from app.services.movie_analyzer_v2 import get_ai_summary
result = await get_ai_summary(mock_llm, None)
assert result is None
mock_llm.ainvoke.assert_not_called()
# ... (Des tests similaires pourraient être écrits pour get_ai_opinion_summary,
# get_ai_best_genre, et get_ai_tags)
# --- Test du service principal (V2) ---
@pytest.mark.asyncio
async def test_service_v2_analyze_movie_partial_request(mocker, mock_llm, mock_movie):
"""
Teste analyze_movie V2 avec une demande partielle (juste aiSummary).
Vérifie que SEULS les appels nécessaires sont faits.
"""
# 1. Mocker les dépendances (repositories)
mock_movie_repo = AsyncMock()
mock_movie_repo.find_by_id.return_value = mock_movie
mocker.patch('app.services.movie_analyzer_v2.movie_repository', mock_movie_repo)
mock_genre_repo = AsyncMock()
mocker.patch('app.services.movie_analyzer_v2.genre_repository', mock_genre_repo)
# 2. Mocker les helpers LLM (pour les espionner)
mock_summary = AsyncMock(return_value="Résumé IA")
mock_opinion = AsyncMock()
mock_genre = AsyncMock()
mock_tags = AsyncMock()
mocker.patch('app.services.movie_analyzer_v2.get_ai_summary', mock_summary)
mocker.patch('app.services.movie_analyzer_v2.get_ai_opinion_summary', mock_opinion)
mocker.patch('app.services.movie_analyzer_v2.get_ai_best_genre', mock_genre)
mocker.patch('app.services.movie_analyzer_v2.get_ai_tags', mock_tags)
# 3. Appel du service
from app.services.movie_analyzer_v2 import analyze_movie
result = await analyze_movie(
movie_id="1",
ai_summary=True,
ai_opinion_summary=False,
ai_best_genre=False,
ai_tags=False,
llm=mock_llm
)
# 4. Assertions
# Le repo de film a été appelé
mock_movie_repo.find_by_id.assert_called_once_with("1")
# Le repo de genre NE DOIT PAS être appelé
mock_genre_repo.list.assert_not_called()
# Seul le helper de résumé a été appelé
mock_summary.assert_called_once()
mock_opinion.assert_not_called()
mock_genre.assert_not_called()
mock_tags.assert_not_called()
# Le résultat est correct
assert result['id'] == strawberry.ID("1")
assert result['aiSummary'] == "Résumé IA"
assert result['aiOpinionSummary'] is None
@pytest.mark.asyncio
async def test_service_v2_analyze_movie_full_request(mocker, mock_llm, mock_movie, mock_genres):
"""
Teste analyze_movie V2 avec une demande complète.
Vérifie que tout est appelé (y compris asyncio.gather).
"""
# 1. Mocker les dépendances (repositories)
mock_movie_repo = AsyncMock()
mock_movie_repo.find_by_id.return_value = mock_movie
mocker.patch('app.services.movie_analyzer_v2.movie_repository', mock_movie_repo)
mock_genre_repo = AsyncMock()
mock_genre_repo.list.return_value = mock_genres
mocker.patch('app.services.movie_analyzer_v2.genre_repository', mock_genre_repo)
# 2. Mocker les helpers LLM
mocker.patch('app.services.movie_analyzer_v2.get_ai_summary', AsyncMock(return_value="Résumé IA"))
mocker.patch('app.services.movie_analyzer_v2.get_ai_opinion_summary', AsyncMock(return_value="Opinion IA"))
mocker.patch('app.services.movie_analyzer_v2.get_ai_best_genre', AsyncMock(return_value="Genre IA"))
mocker.patch('app.services.movie_analyzer_v2.get_ai_tags', AsyncMock(return_value=["tag1", "tag2"]))
# 3. Appel du service
from app.services.movie_analyzer_v2 import analyze_movie
result = await analyze_movie(
movie_id="1",
ai_summary=True,
ai_opinion_summary=True,
ai_best_genre=True,
ai_tags=True,
llm=mock_llm
)
# 4. Assertions
# Les deux repos ont été appelés
mock_movie_repo.find_by_id.assert_called_once_with("1")
mock_genre_repo.list.assert_called_once() # Appelé car ai_best_genre=True
# Le résultat est complet
assert result['aiSummary'] == "Résumé IA"
assert result['aiOpinionSummary'] == "Opinion IA"
assert result['aiBestGenre'] == "Genre IA"
assert result['aiTags'] == ["tag1", "tag2"]

110
tests/etape6_resolver_v2.py Normal file
View File

@@ -0,0 +1,110 @@
"""
Tests de validation pour l'Étape 6 : Optimisation du Resolver GraphQL (Version 2)
Objectif :
1. Vérifier que le resolver V2 récupère bien le LLM du contexte.
2. Vérifier que le resolver V2 utilise `is_field_requested` pour
passer les bons booléens au service V2.
Prérequis :
- Les TODOs de `app/graphql/resolvers/analyze_movie_v2.py` sont complétés.
- Le champ `analyzeMovie` de `app/graphql/queries.py` est mis à jour
pour utiliser le resolver V2.
"""
import pytest
import strawberry
from unittest.mock import AsyncMock, MagicMock, patch
@pytest.fixture
def mock_info():
"""Fixture pour un objet Info de Strawberry."""
mock_llm_instance = MagicMock(name="MockLLM")
info = MagicMock()
info.context = {"llm": mock_llm_instance}
return info
@pytest.mark.asyncio
async def test_resolver_v2_partial_request(mocker, mock_info):
"""
Teste le resolver V2 avec une requête partielle.
Vérifie qu'il passe les bons drapeaux au service.
"""
# 1. Mocker les dépendances (le service V2 et le helper is_field_requested)
# On mock le service V2 pour espionner ses arguments
mock_service = AsyncMock(return_value={
"id": "1",
"aiSummary": "Service Result",
"aiOpinionSummary": None,
"aiBestGenre": None,
"aiTags": None
})
mocker.patch('app.graphql.resolvers.analyze_movie_v2.analyze_movie', mock_service)
# On mock 'is_field_requested' pour simuler une requête partielle
def mock_is_field_requested(info, field_name):
if field_name == "aiSummary":
return True
return False
mocker.patch('app.graphql.resolvers.analyze_movie_v2.is_field_requested', mock_is_field_requested)
# 2. Appel du resolver
from app.graphql.resolvers.analyze_movie_v2 import analyze_movie_by_id
await analyze_movie_by_id(
movie_id=strawberry.ID("1"),
info=mock_info
)
# 3. Assertions
# Vérifie que le service a été appelé avec les bons drapeaux
mock_service.assert_called_once_with(
movie_id="1",
ai_summary=True,
ai_opinion_summary=False,
ai_best_genre=False,
ai_tags=False,
llm=mock_info.context["llm"] # Vérifie que le LLM du contexte est bien passé
)
@pytest.mark.asyncio
async def test_resolver_v2_full_request(mocker, mock_info):
"""
Teste le resolver V2 avec une requête complète.
"""
# 1. Mocker les dépendances
# [CORRECTION] Le mock doit retourner un dictionnaire complet
# pour que MovieAnalysis(**analysis_data) fonctionne.
mock_service_return = {
"id": "1",
"aiSummary": "Mock summary",
"aiOpinionSummary": "Mock opinion",
"aiBestGenre": "Mock genre",
"aiTags": ["mock", "tag"]
}
mock_service = AsyncMock(return_value=mock_service_return)
mocker.patch('app.graphql.resolvers.analyze_movie_v2.analyze_movie', mock_service)
# Simule une requête complète
mocker.patch('app.graphql.resolvers.analyze_movie_v2.is_field_requested', return_value=True)
# 2. Appel du resolver
from app.graphql.resolvers.analyze_movie_v2 import analyze_movie_by_id
await analyze_movie_by_id(
movie_id=strawberry.ID("1"),
info=mock_info
)
# 3. Assertions
# Vérifie que le service a été appelé avec TOUS les drapeaux à True
mock_service.assert_called_once_with(
movie_id="1",
ai_summary=True,
ai_opinion_summary=True,
ai_best_genre=True,
ai_tags=True,
llm=mock_info.context["llm"]
)

51
tests/etape7_erreurs.py Normal file
View File

@@ -0,0 +1,51 @@
"""
Tests de validation pour l'Étape 7 (Bonus) : Gestion fine des erreurs métier
Objectif :
1. Vérifier que le service V2 lève bien une `NotFoundBLLException`
si le `movie_repository` retourne `None`.
Prérequis :
- La logique de gestion d'erreur est en place dans
`app/services/movie_analyzer_v2.py`.
"""
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.core.exceptions import NotFoundBLLException
@pytest.mark.asyncio
async def test_service_v2_raises_not_found(mocker):
"""
Vérifie que le service `analyze_movie` lève une `NotFoundBLLException`
si le film n'est pas trouvé dans le repository.
"""
# 1. Mocker les dépendances (repositories)
mock_movie_repo = AsyncMock()
mock_movie_repo.find_by_id.return_value = None # Simule un film non trouvé
mocker.patch('app.services.movie_analyzer_v2.movie_repository', mock_movie_repo)
mock_llm = MagicMock()
# 2. Appel du service en s'attendant à une exception
from app.services.movie_analyzer_v2 import analyze_movie
with pytest.raises(NotFoundBLLException) as exc_info:
await analyze_movie(
movie_id="999",
ai_summary=True, # Peu importe les drapeaux
ai_opinion_summary=False,
ai_best_genre=False,
ai_tags=False,
llm=mock_llm
)
# 3. Assertions
# Vérifie que le repo a bien été appelé
mock_movie_repo.find_by_id.assert_called_once_with("999")
# Vérifie que le type d'exception est correct
assert exc_info.type is NotFoundBLLException
# Vérifie que le message d'erreur contient l'ID
assert "999" in str(exc_info.value)
assert "Movie" in str(exc_info.value)