Merge for modification and comments

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)

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))