diff --git a/app.py b/app.py index c4ca110..9c1d237 100644 --- a/app.py +++ b/app.py @@ -1,61 +1,118 @@ import asyncio -from typing import Annotated from functools import lru_cache +from fastapi import FastAPI, APIRouter, Depends, HTTPException, Query, Request from dotenv import load_dotenv -from fastapi import FastAPI, APIRouter, Depends, HTTPException, Query from fastapi.security import OAuth2PasswordRequestForm +from settings import Settings +from typing import Annotated from Authentification import User, get_current_active_user, fake_users_db, UserInDB, fake_hash_password, \ get_current_admin_user from pydantic import BaseModel -from settings import Settings +import pickle +from collections import Counter +import os +import time load_dotenv() settings1 = Settings() my_router = APIRouter() -app = FastAPI(title=settings1.title, - description= settings1.description, - ) +app = FastAPI(title=settings1.title, description=settings1.description) +route_request_counter = Counter() +route_time_counter = Counter() -@lru_cache -def get_settings(): - return Settings() +@app.on_event("shutdown") +def shutdown_event(): + """ + Remove the statistics file when the app shutdown + """ + try: + os.remove('saved_count.pkl') + except Exception as e: + print(f"Error deleting statistics file: {e}") -def get_saved_value(): +def get_saved_values(): + """ + Retrieve the statistics data saved in the file + :return: dict with statistics values of all API routes + """ try: - with open("saved_count.txt", "r") as file: - value = int(file.read()) + with open("saved_count.pkl", "rb") as file: + values = pickle.load(file) except FileNotFoundError: - with open("saved_count.txt", 'w') as file: - file.write('0') - value = 0 - return value + with open('saved_count.pkl', 'wb') as file: + values = dict() + pickle.dump(values, file) + return values + + +def save_value(values): + """ + Save the current API statistics in a file + :param values: dict with statistics values of all API routes + """ + with open("saved_count.pkl", "wb") as file: + pickle.dump(values, file) + + +@app.middleware("http") +async def count_requests(request: Request, call_next): + """ + Increment the number of request by route and calculate the time while processing + :param request: incoming HTTP request + :param call_next: function that represents the next middleware or request handler in the processing pipeline + :return: response of the API request + """ + route = request.url.path + route_request_counter[route] += 1 + start_time = time.time() + response = await call_next(request) + end_time = time.time() + route_time_counter[route] += round(end_time - start_time, 6) + return response + + +def check_args_type(dict_args, type_args): + """ + Check if args in the dictionary corresponds to the expected type, raise an error if not + :param dict_args: dict of arguments to check + :param type_args: list of expected types + """ + for value, expected_type in zip(dict_args.values(), type_args): + if not isinstance(value, expected_type): + raise TypeError( + f"Type d'argument incorrect. Attendu : {expected_type.__name__}, Reçu : {type(value).__name__}" + ) -request_count = get_saved_value() +@lru_cache +def get_settings(): + return Settings() -def save_value(value): - with open("saved_count.txt", "w") as file: - file.write(str(value)) +def count_func_call(func): + """ + Increment the number of call by fonction + :param func: methode to increment the count + """ + request_count = get_saved_values() + key_func = func.__name__ + if key_func in request_count: + request_count[key_func] += 1 + else: + request_count[key_func] = 1 + save_value(request_count) def fast_api_decorator(route, method, type_args): def decorator(func): def wrapper(**kwargs): - # Handle argument type error if type_args is not None - if type_args is not None: - for value, expected_type in zip(kwargs.values(), type_args): - if not isinstance(value, expected_type): - raise TypeError(f"Type d'argument incorrect. Attendu : {expected_type.__name__}, Reçu : {type(value).__name__}") - + # Handle argument type error + check_args_type(dict_args=kwargs, type_args=type_args) # Count the number of request - global request_count - request_count += 1 - save_value(request_count) - + count_func_call(func=func) # add endpoint to the API my_router.add_api_route(path=route, endpoint=func, methods=method) app.include_router(my_router) @@ -64,29 +121,57 @@ def wrapper(**kwargs): return decorator +@fast_api_decorator(route="/power/", method=["GET"], type_args=[int, int]) +def power_function(x: Annotated[int, Query(description="Int we'll compute the power")], + a: Annotated[int, Query(description="Power of the calculation")], + current_user: User = Depends(get_current_active_user)): + return {f"{x} to the power of {a}": x ** a} + + @fast_api_decorator(route="/add/", method=["GET"], type_args=[int, int]) -def add_function(x: Annotated[int, Query(description="Int we'll add something")], a: Annotated[int, Query(description="Int added")],current_user: User = Depends(get_current_active_user)): +def add_function(x: Annotated[int, Query(description="Int we'll add something")], + a: Annotated[int, Query(description="Int added")], + current_user: User = Depends(get_current_active_user)): return {f"{x} + {a} equals": x + a} @fast_api_decorator(route="/sous/", method=["GET"], type_args=[int, list]) -def sous_function(x: Annotated[int, Query(description="Int we'll substract something")], lst: Annotated[list[int], Query(description="List of 2 int that will be substracted")],current_user: User = Depends(get_current_active_user)): +def sous_function(x: Annotated[int, Query(description="Int we'll substract something")], + lst: Annotated[list[int], Query(description="List of 2 int that will be substracted")], + current_user: User = Depends(get_current_active_user)): return {f"{x} - {lst[0]} - {lst[1]} equals": x - lst[0] - lst[1]} -@fast_api_decorator(route="/users/me", method=["GET"], type_args=None) -def read_users_me(current_user: User = Depends(get_current_active_user)): - return current_user +class InputDiv(BaseModel): + div: int +# Pour faire une requête avec un argument "Body" ou un json avec des arguments il faut passer +# par une méthode "POST" et pas "GET" +@fast_api_decorator(route="/div/", method=["POST"], type_args=[int, InputDiv]) +def div_function(x: Annotated[int, Query(description="Int we will divide something")], item: InputDiv): + return {f"{x} / {item.div} equals": item.div} -@fast_api_decorator(route="/power/", method=["POST"],type_args=[int, int]) -def power_function(x: Annotated[int, Query(description="Int we'll add something")], a: Annotated[int, Query(description="Int added")], current_user: User = Depends(get_current_active_user)): - return {f"{x} to the power of {a}": int(x)**int(a)} + +@app.get("/stats") +async def get_stats(current_user: User = Depends(get_current_admin_user)): + avg_time = dict() + for key in route_request_counter.keys(): + avg_time[key] = route_time_counter[key] * 1000 / route_request_counter[key] + return {"Nombre d'appels aux fonctions décorées": get_saved_values(), + "Nombre d'appels totaux des API par route": route_request_counter, + "Temps moyen d'exécution par route en ms": avg_time} + + +@fast_api_decorator(route="/users/me", method=["GET"], type_args=[]) +def read_users_me(current_user: User = Depends(get_current_active_user)): + return current_user @fast_api_decorator(route="/rendement/", method=["POST"], type_args=[int, float]) -def rendement(x: Annotated[int, Query(description="Int we'll add something")], r: Annotated[float, Query(description="float added")], current_user: User = Depends(get_current_active_user)): +def rendement(x: Annotated[int, Query(description="Int we'll add something")], + r: Annotated[float, Query(description="float added")], + current_user: User = Depends(get_current_active_user)): return {f"{x} * (1 + {r}) equals": int(x) * (1 + float(r))} @@ -99,7 +184,6 @@ async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]): hashed_password = fake_hash_password(form_data.password) if not hashed_password == user.hashed_password: raise HTTPException(status_code=400, detail="Incorrect username or password") - return {"access_token": user.username, "token_type": "bearer"} @@ -112,23 +196,6 @@ async def info(): } -class InputDiv(BaseModel): - div: int - - -# Pour faire une requête avec un argument "Body" ou un json avec des arguments il faut passer -# par une méthode "POST" et pas "GET" -@fast_api_decorator(route="/div/", method=["POST"], type_args=[int, InputDiv]) -def div_function(x: Annotated[int, Query(description="Int we will divide something")], item: InputDiv,current_user: User = Depends(get_current_active_user)): - return {f"{x} / {item.div} equals": item.div} - - -@app.get("/stats") -async def get_stats(current_user: User = Depends(get_current_admin_user)): - request_count = get_saved_value() - return {"request_count": request_count} - - # On "lance" les fonctions pour qu'elles soient visibles par l'app FastAPI read_users_me() rendement(x=0, r=0.0) diff --git a/decorator.py b/decorator.py deleted file mode 100644 index 3e40809..0000000 --- a/decorator.py +++ /dev/null @@ -1,18 +0,0 @@ -# In this script we build the decorator and the function we want to transform into an API -from app import app, my_router - - -def fast_api_decorator(route): - def decorator(func): - def wrapper(x, a): - func(x, a) - return func(x, a) - my_router.add_api_route(path='/{func}/{x}/{a}', endpoint=wrapper) - app.include_router(my_router) - return wrapper - return decorator - - -@fast_api_decorator(route="") -def power_function(x, a): - return {f"{x} to the power of {a}": x**a} diff --git a/main.py b/main.py index 5a4c9e5..d04719b 100644 --- a/main.py +++ b/main.py @@ -1,27 +1,5 @@ -""" -FastApiDecoratorBuilder - -Description du Projet : -Votre défi dans le projet "FastApiDecoratorBuilder" est de concevoir un décorateur Python qui -transforme une fonction Python en une API FastAPI basée sur la fonction et des configurations -définies. - -Objectifs du Projet : -- Création de Décorateur: Construire un décorateur qui transforme une fonction Python en API FastAPI. -- Gestion de Configurations: Implanter un mécanisme de configuration pour l’API. - -Consignes : -1) Développement du Décorateur : Elaborez un décorateur qui, appliqué à une fonction, génère une API FastAPI -correspondante. -2) Configuration de l'API : Intégrez une méthode pour configurer les propriétés de l’API générée, telles que -les routes et les méthodes HTTP acceptées. -""" - -# Use command : "python -m uvicorn main:app --reload" to lauch server and be able to request the "app" API. - from app import app, power_function, add_function, sous_function, rendement import requests -import asyncio def power(x, a): @@ -40,7 +18,7 @@ def sous(x, lst): power(x=9, a=2) add(x=9, a=2) sous(x=9, lst=[2, 1]) - print(rendement(x=0, r=0.0)) print(power_function(x=9, a=2)) print(add_function(x=9, a=2)) print(sous_function(x=9, lst=[2, 1])) + print(rendement(x=0, r=0.0))