Merge pull request #2 from Minibrams/feature/function-wrappers

Added support for overriding endpoint functions while using add_dependencies()

Author: Minibrams

README.md

@@ -11,12 +11,41 @@ Create decorators that leverage FastAPI's `Depends()` and built-in dependencies,
 pip install fastapi-decorators
 ```
 
+# TL;DR
+The library supplies the `add_dependencies()` decorator function which effectively allows you to add argument dependencies to your FastAPI endpoints.
+
+For example, the following three endpoints have the same signature:
+```python
+# Using normal dependencies
+@app.get("/items/{item_id}")
+def read_item(item_id: int, _ = Depends(get_current_user)):
+    ...
+
+# Using add_dependency directly
+@app.get("/items/{item_id}")
+@add_dependencies(Depends(get_current_user))
+def read_item(item_id: int):
+    ...
+
+# Using a custom decorator
+def authorize():
+    def dependency(user = Depends(get_current_user)):
+        return user
+    return add_dependencies(Depends(dependency))
+
+@app.get("/items/{item_id}")
+@authorize()
+def read_item(item_id: int):
+    ...
+```
+
 # Usage examples
 
 - [Logging decorator](#logging-decorator)
 - [Authorization decorator](#authorization-decorator)
 - [Custom Response Header decorator](#custom-response-header-decorator)
 - [Rate Limiting decorator](#rate-limiting-decorator)
+- [Caching decorator](#caching-decorator)
 - [Error Handling decorator](#error-handling-decorator)
 - [Combining Multiple decorators](#combining-multiple-decorators)
 - [Using `add_dependencies()` directly](#using-add_dependencies-directly)
@@ -126,26 +155,70 @@ def limited_endpoint():
 
 ```
 
+## Caching decorator
+Add caching to your endpoints:
+
+```python
+def get_cache() -> dict:
+    return {}  # Use a real cache like Redis or Memcached
+
+def cache_response(max_age: int = 5):
+    def decorator(func):
+
+        # Wrap the endpoint after adding the get_cache dependency
+        @add_dependencies(cache=Depends(get_cache))
+        @wraps(func)
+        def wrapper(*args, cache: dict, **kwargs):
+            key = func.__name__
+
+            if key in cache:
+                timestamp, data = cache[key]
+                if time() - timestamp < max_age:
+                    # Cache hit
+                    return data
+
+            # Cache miss - call the endpoint as usual
+            result = func(*args, **kwargs)
+
+            # Store the result in the cache
+            cache[key] = time(), result
+            return result
+        
+        return wrapper
+    return decorator
+
+@app.get("/cached-data")
+@cache_response(max_age=10)
+def get_cached_data():
+    ...
+```
+
 ## Error Handling decorator
 Create a decorator to handle exceptions and return custom responses:
 
 ```python
 from fastapi_decorators import add_dependencies
 from fastapi import Depends, Response
-import traceback
-
-def handle_errors():
-    async def dependency(response: Response):
-        try:
-            yield
-        except Exception as e:
-            response.status_code = 500
-            response.content = f"An error occurred: {str(e)}"
 
-            # Optionally print the traceback
-            traceback.print_exc()
+def get_crash_log_storage() -> list:
+    return []  # Use a real storage like a database
 
-    return add_dependencies(Depends(dependency))
+def handle_errors():
+    def decorator(func):
+
+        # Wrap the endpoint after adding the crash_logs dependency
+        @add_dependencies(crash_logs = Depends(get_crash_log_storage))
+        @wraps(func)
+        def wrapper(*args, crash_logs: list, **kwargs):
+            try:
+                return func(*args, **kwargs)
+            except Exception as e:
+                # Log the error and return a custom response
+                crash_logs.append({ 'error': str(e), 'function': func.__name__ })
+                return JSONResponse(status_code=500, content={ "detail": str(e) })
+            
+        return wrapper
+    return decorator
 
 @app.get("/may-fail")
 @handle_errors()

setup.py

@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="fastapi-decorators",
-    version="1.0.2",
+    version="1.0.3",
     author="Anders Brams",
     author_email="anders@brams.dk",
     description="Create decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.",

src/fastapi_decorators/decorators.py

@@ -4,11 +4,11 @@
 from functools import wraps
 from inspect import Parameter, signature
 from types import MappingProxyType
-from typing import Any, Callable, Dict, Tuple, TypeVar, cast
+from typing import Any, Callable, Tuple, TypeVar, cast
 
 F = TypeVar('F', bound=Callable[..., Any])
 
-def add_dependencies(*dependencies: Any) -> Callable[[F], F]:
+def add_dependencies(*args: Any, **kwargs: Any) -> Callable[[F], F]:
     """
     Decorator to add dependencies to a function without exposing them as arguments.
 
@@ -50,10 +50,10 @@ def decorator(func: F) -> F:
         original_signature = signature(func)
         original_parameters = original_signature.parameters
 
-        new_parameters = _add_dependency_parameters(dependencies, original_parameters)
+        new_parameters = _add_dependency_parameters(args, kwargs, original_parameters)
         new_signature = original_signature.replace(parameters=tuple(new_parameters.values()))
 
-        wrapper = _create_wrapper(func, original_parameters)
+        wrapper = _create_wrapper(func, new_parameters)
         wrapper.__signature__ = new_signature  # type: ignore
 
         return cast(F, wrapper)
@@ -62,8 +62,9 @@ def decorator(func: F) -> F:
 
 def _add_dependency_parameters(
     dependencies: Tuple[Any, ...],
+    named_dependencies: dict[str, Any],
     original_parameters: MappingProxyType[str, Parameter],
-) -> Dict[str, Parameter]:
+) -> dict[str, Parameter]:
     """
     Adds dependency parameters to the function's parameters.
 
@@ -84,12 +85,19 @@ def _add_dependency_parameters(
             annotation=Any,
         )
 
-    return new_parameters
+    for name, dependency in named_dependencies.items():
+        new_parameters[name] = Parameter(
+            name,
+            kind=Parameter.KEYWORD_ONLY,
+            default=dependency,
+            annotation=Any,
+        )
 
+    return new_parameters
 
 def _generate_dependency_name(
     index: int, 
-    current_parameters: Dict[str, Parameter],
+    current_parameters: dict[str, Parameter],
 ) -> str:
     """
     Generates a unique name for an anonymous dependency.
@@ -112,7 +120,7 @@ def _generate_dependency_name(
 
 def _create_wrapper(
     func: Callable, 
-    original_parameters: MappingProxyType[str, Parameter],
+    original_parameters: dict[str, Parameter],
 ) -> Callable:
     """
     Creates a wrapper function that filters out dependency arguments.
@@ -128,15 +136,15 @@ def _create_wrapper(
         @wraps(func)
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
             filtered_kwargs = {
-                k: v for k, v in kwargs.items() if k in original_parameters
+                k: v for k, v in kwargs.items() if k in original_parameters and not k.startswith("__dependency_")
             }
             return await func(*args, **filtered_kwargs)
         return async_wrapper
     else:
         @wraps(func)
         def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
             filtered_kwargs = {
-                k: v for k, v in kwargs.items() if k in original_parameters
+                k: v for k, v in kwargs.items() if k in original_parameters and not k.startswith("__dependency_")
             }
             return func(*args, **filtered_kwargs)
         return sync_wrapper

tests/app.py

@@ -1,10 +1,10 @@
 from functools import wraps
+from inspect import signature
 import logging
-from time import time
+from time import sleep, time
 from fastapi import Depends, FastAPI, HTTPException, Header, Request, Response
 from fastapi.responses import JSONResponse
 from pydantic import BaseModel
-from requests import Session
 
 from fastapi_decorators import add_dependencies
 
@@ -18,6 +18,7 @@ class DataModel(BaseModel):
     value: int
 
 rate_limit_store = {}
+cache_storage = {}
 error_log = []
 fake_db = {
     "access_token": "valid_token",
@@ -36,6 +37,12 @@ class DataModel(BaseModel):
 def get_db() -> dict:
     return fake_db
 
+def get_cache() -> dict:
+    return cache_storage
+
+def get_crash_log_storage() -> list:
+    return error_log
+
 def get_rate_limit_store() -> dict:
     return rate_limit_store
 
@@ -98,17 +105,42 @@ async def dependency(
         rate_limit_store[request_id] = calls_info
     return add_dependencies(Depends(dependency))
 
+def cache_response(max_age: int = 5):
+    def decorator(func):
+
+        # Wrap the endpoint after adding the get_cache dependency
+        @add_dependencies(cache=Depends(get_cache))
+        @wraps(func)
+        def wrapper(*args, cache: dict, **kwargs):
+            key = func.__name__
+
+            if key in cache:
+                timestamp, data = cache[key]
+                if time() - timestamp < max_age:
+                    # Cache hit
+                    return data
+
+            # Cache miss - call the endpoint as usual
+            result = func(*args, **kwargs)
+
+            # Store the result in the cache
+            cache[key] = time(), result
+            return result
+        
+        return wrapper
+    return decorator
+
 def handle_errors():
     def decorator(func):
+        @add_dependencies(crash_logs = Depends(get_crash_log_storage))
         @wraps(func)
-        async def wrapper(*args, **kwargs):
+        def wrapper(*args, crash_logs: list, **kwargs):
             try:
-                return await func(*args, **kwargs)
+                return func(*args, **kwargs)
             except Exception as e:
-                error_message = f"An error occurred: {str(e)}"
-                logging.error(error_message)
-                error_log.append({'error': error_message, 'function': func.__name__})
-                return JSONResponse(status_code=500, content={"detail": error_message})
+                crash_logs.append({ 'error': str(e), 'function': func.__name__ })
+                return JSONResponse(status_code=500, content={ "detail": str(e) })
+            
         return wrapper
     return decorator
 
@@ -185,9 +217,18 @@ def limited_endpoint():
     """
     return {"message": "You have accessed a rate-limited endpoint"}
 
+@app.get("/expensive-operation")
+@cache_response(max_age=5)
+def expensive_operation():
+    """
+    Endpoint that is cached for 5 seconds.
+    """
+    sleep(5)
+    return {"data": time() }
+
 @app.get("/may-fail")
 @handle_errors()
-async def may_fail_operation(should_fail: bool = False):
+def may_fail_operation(should_fail: bool = False):
     """
     Endpoint that may raise exceptions.
     """

tests/test_app.py

@@ -1,6 +1,6 @@
-from time import sleep
+from time import sleep, time
 from fastapi.testclient import TestClient
-from app import app, fake_db, rate_limit_store, error_log
+from app import app, fake_db, rate_limit_store, cache_storage, error_log
 
 client = TestClient(app)
 
@@ -99,6 +99,22 @@ def test_rate_limiting():
     response = client.get("/limited-endpoint")
     assert response.status_code == 200
 
+def test_cached_response():
+    cache_storage.clear()
+    data1 = client.get("/expensive-operation").json()
+    start = time()
+    data2 = client.get("/expensive-operation").json()
+    end = time()
+
+    assert end - start < 1  # The endpoint takes 5 seconds when not cached
+    assert data1 == data2
+
+    # Wait for the cache to expire
+    sleep(5)
+
+    data3 = client.get("/expensive-operation").json()
+    assert data1 != data3
+
 def test_error_handling_success():
     response = client.get("/may-fail?should_fail=false")
     assert response.status_code == 200
@@ -109,7 +125,7 @@ def test_error_handling_failure():
 
     response = client.get("/may-fail?should_fail=true")
     assert response.status_code == 500
-    assert "An error occurred" in response.json()["detail"]
+    assert len(error_log) == 1
 
     # Check that error was logged
     assert len(error_log) == 1