api-avr23-cds-translation1

Sleeping

App Files Files Community

api-avr23-cds-translation1 / main.py

Demosthene-OR

Update main.py

3d78a2b 3 months ago

raw history blame contribute delete

No virus

7.67 kB

	from fastapi import FastAPI, HTTPException, Header, Depends, Request
	from fastapi.responses import JSONResponse
	from fastapi.security import HTTPBasic, HTTPBasicCredentials
	from fastapi.exceptions import RequestValidationError
	from typing import Optional, List
	from pydantic import BaseModel, ValidationError
	import pandas as pd
	import datetime

	api = FastAPI()

	# Charger les données à partir du fichier CSV
	questions_data = pd.read_csv('questions.csv')

	# Dictionnaire des identifiants des utilisateurs
	users_credentials = {
	"alice": "wonderland",
	"bob": "builder",
	"clementine": "mandarine",
	"admin": "4dm1N" # Ajout de l'utilisateur admin
	}

	# Modèle Pydantic pour représenter une question
	class Question(BaseModel):
	question: str
	subject: str
	correct: Optional[str] = None # Champ optionnel
	use: str
	responseA: str
	responseB: str
	responseC: Optional[str] = None # Champ optionnel
	responseD: Optional[str] = None # Champ optionnel
	# remark: Optional[str] = None # Champ optionnel

	# Modèle pour représenter une exception personnalisée
	class MyException(Exception):
	def __init__(self,
	status_code: int,
	name : str,
	message : str):
	self.status_code = status_code
	self.name = name
	self.message = message
	self.date = str(datetime.datetime.now())

	# Gestionnaire d'exception personnalisé
	@api.exception_handler(MyException)
	def MyExceptionHandler(
	request: Request,
	exception: MyException
	):
	return JSONResponse(
	status_code=exception.status_code,
	content={
	'url': str(request.url),
	'name': exception.name,
	'message': exception.message,
	'date': exception.date
	}
	)

	# Gestionnaire d'exception pour les erreurs de validation de la requête
	@api.exception_handler(RequestValidationError)
	async def validation_exception_handler(request: Request, exc: RequestValidationError):
	return JSONResponse(
	status_code=422,
	content={
	'url': str(request.url),
	'name': "Erreur de validation de la requête (parametre requis)",
	'message': exc.errors(),
	'date': str(datetime.datetime.now())
	},
	)

	# Gestionnaire d'exception pour les erreurs Pydantic
	@api.exception_handler(ValidationError)
	async def validation_exception_handler(request: Request, exc: ValidationError):
	return JSONResponse(
	status_code=422,
	content={
	'url': str(request.url),
	'name': "Erreur de validation Pydantic",
	'message': exc.errors(),
	'date': str(datetime.datetime.now())
	},
	)

	# Fonction pour vérifier l'authentification de l'utilisateur
	def authenticate(authorisation: str = Header(None)):
	if not authorisation:
	raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
	try:
	scheme, credentials = authorisation.split()
	if scheme != 'Basic':
	raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
	username, password = credentials.split(":")
	if username not in users_credentials or users_credentials[username] != password:
	raise HTTPException(status_code=401, detail="Utilisateur non authorisé ")
	except Exception as e:
	raise HTTPException(status_code=401, detail="Utilisateur non authorisé")
	return username, password

	# Endpoint pour vérifier que l'API est fonctionnelle
	@api.get('/', name="Vérification que l'API fonctionne")
	def check_api():
	return {'message': "L'API fonctionne"}

	# Endpoint pour récupérer les questions en fonction du type de test (use) et des catégories (subject) spécifiés
	@api.get('/questions', name="Récupération des questions pour un QCM")
	def get_questions(use: str,
	subject: str,
	num_questions: int,
	authorisation: str = Header(None)):
	"""
	Récupère les questions en fonction du type de test (use) et des catégories (subject) spécifiés
	L'application peut produire des QCMs de 5, 10 ou 20 questions (seulement)
	Les questions sont retournées dans un ordre aléatoire
	Seuls les utilisateurs se trouvant dans users_credentials peuvent utiliser cette application
	"""

	# Verifier si l'utilsateur existe et a le droit
	authenticate(authorisation)

	# Verifier si le nombre de questions demandé correspond au nombre de questions d'un QCM
	if num_questions not in [5,10,20]:
	raise MyException(status_code=422, name="num_questions invalide", \
	message="La requête peut contenir 51,10 ou 20 questions, mais pas "+str(num_questions))

	# Filtrer les questions en fonction des paramètres spécifiés
	filtered_questions = questions_data
	if use:
	filtered_questions = filtered_questions[filtered_questions['use'] == use]
	if subject is not None:
	s = subject.split(',')
	filtered_questions = filtered_questions[filtered_questions['subject'].isin(s)]
	print("len(filtered_questions)=",len(filtered_questions))
	print("num_questions=",num_questions)
	# Vérifier si des questions sont disponibles dans la catégorie spécifiée
	if len(filtered_questions) == 0:
	raise MyException(status_code=404, name="Aucune question", \
	message="Aucune question ne correspond aux critères sélectionnés")
	# Verifier si le nombre de questions diponibles >= au nombre requis
	elif (len(filtered_questions) < num_questions):
	raise MyException(status_code=400, name="Nb insuffisant de questions ", \
	message="Le nombre de questions correspondantes aux critères sélectionnés est < au nombre requis")

	# Supprimer les valeurs NaN dans les champs correct,responseC, responseD (ils ne sont pas toujours remplis)
	filtered_questions['correct'] = filtered_questions['correct'].apply(lambda x: None if pd.isna(x) else x)
	filtered_questions['responseC'] = filtered_questions['responseC'].apply(lambda x: None if pd.isna(x) else x)
	filtered_questions['responseD'] = filtered_questions['responseD'].apply(lambda x: None if pd.isna(x) else x)

	# Sélectionner un nombre aléatoire de questions
	selected_questions = filtered_questions.sample(n=min(num_questions, len(filtered_questions)))

	# Convertir les données en liste de dictionnaires
	questions_list = selected_questions.to_dict(orient='records')

	# Convertir les dictionnaires en objets Pydantic de type Question
	questions_objects = [Question(**question) for question in questions_list]
	return questions_objects


	# Endpoint pour créer une nouvelle question (accessible uniquement par l'utilisateur admin)
	@api.post('/questions/create', name="Création d'une nouvelle question")
	def create_question(question: Question,
	authorisation: str = Header(None)):
	"""
	Crée une nouvelle question et l'ajoute à questions.csv
	Seuls l' utilisateur admin a le droit d'utiliser cette fonction
	"""
	global questions_data

	username, password = authenticate(authorisation)
	if username != 'admin':
	raise HTTPException(status_code=401, detail="Utilisateur non authorisé")

	# Ajouter la nouvelle question au DataFrame
	new_question = question.model_dump()
	questions_data = questions_data.append(new_question, ignore_index=True)

	# Sauvegarder les modifications dans le fichier CSV
	questions_data.to_csv('questions.csv', index=False)

	return {'message': 'Question créée avec succès'}