fastapi-decorators


Namefastapi-decorators JSON
Version 1.0.4 PyPI version JSON
download
home_pagehttps://github.com/Minibrams/fastapi-decorators
SummaryCreate decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.
upload_time2024-10-03 00:36:03
maintainerNone
docs_urlNone
authorAnders Brams
requires_python>=3
licenseNone
keywords fastapi decorators middleware dependency dependencies
VCS
bugtrack_url
requirements No requirements were recorded.
Travis-CI No Travis.
coveralls test coverage No coveralls.
            # FastAPI decorators
![CI](https://github.com/Minibrams/fastapi-decorators/actions/workflows/ci.yml/badge.svg)
[![PyPI](https://img.shields.io/pypi/v/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)
[![Python Versions](https://img.shields.io/pypi/pyversions/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)
[![Downloads](https://img.shields.io/pypi/dm/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)

Create decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.

# Installation
```bash
pip install fastapi-decorators
```

# TL;DR
The library supplies the `depends()` 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}")
@depends(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 depends(Depends(dependency))

@app.get("/items/{item_id}")
@authorize()
def read_item(item_id: int):
    ...
```

# Usage examples

- [Using `depends()` directly](#using-depends-directly)
- [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)
- [Dependency injection with parameters](#dependency-injection-with-parameters)

## Using `depends()` directly
If you prefer, you can use depends directly without creating a custom decorator:

```python
from fastapi_decorators import depends
from fastapi import Depends, Header

async def verify_api_key(x_api_key: str = Header(...)):
    if x_api_key != "expected-api-key":
        raise HTTPException(status_code=403, detail="Forbidden")

@app.get("/secure-data")
@depends(Depends(verify_api_key))
def get_secure_data():
    ...

```

## Logging decorator
Add a decorator to log incoming requests:

```python
from fastapi_decorators import depends
from fastapi import Request, Depends

def log_request():
    def dependency(request: Request):
        print(f"Received request: {request.method} {request.url}")
    return depends(Depends(dependency))

@app.get("/items/{item_id}")
@log_request()
def read_item(item_id: int):
    ...

```

## Authorization decorator
Create a simple decorator that rejects unauthorized requests:
> The API docs will reflect the authentication requirement for this endpoint
> because of the added OAuth2 dependency.

```python
from fastapi_decorators import depends
from fastapi import Depends, HTTPException, Header
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.orm import Session

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def authorize(*required_scopes: str):
    def dependency(token: str = Depends(oauth2_scheme)):
        # Your auth logic here
        if not token:
            raise HTTPException(status_code=401, detail="Unauthorized")

        # Check scopes and permissions
        pass
    return depends(Depends(dependency))


@app.put("/users/{user_id}")
@authorize("users:write")
def update_user(*, user_id: int, user_update: UserUpdate):
    ...

```

## Custom Response Header decorator
Create a decorator to add custom headers to responses:

```python
from fastapi_decorators import depends
from fastapi import Response, Depends

def add_custom_header(name: str, value: str):
    def dependency(response: Response):
        response.headers[name] = value
    return depends(Depends(dependency))

@app.get("/data")
@add_custom_header("X-Custom-Header", "MyValue")
def get_data():
    ...

```

## Rate Limiting decorator
Add rate limiting to your endpoints:

```python
from fastapi_decorators import depends
from fastapi import Depends, HTTPException, Request
from time import time

rate_limit_store = {}

def rate_limit(max_calls: int, period: int):
    def dependency(ip_address: str = Depends(get_ip_address)):
        # Simple rate limiting logic
        now = time()
        calls, last_reset = rate_limit_store.get(ip_address, (0, now))
        if now - last_reset > period:
            # Reset rate limit
            calls = 0
            last_reset = now
        if calls >= max_calls:
            raise HTTPException(status_code=429, detail="Too Many Requests")
        calls += 1
        rate_limit_store[ip_address] = (calls, last_reset)
    return depends(Depends(dependency))

def get_ip_address(request: Request):
    return request.client.host

@app.get("/limited-endpoint")
@rate_limit(max_calls=5, period=60)
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
        @depends(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 depends
from fastapi import Depends, Response

def get_crash_log_storage() -> list:
    return []  # Use a real storage like a database

def handle_errors():
    def decorator(func):

        # Wrap the endpoint after adding the crash_logs dependency
        @depends(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()
def may_fail_operation():
    ...

```

## Combining Multiple decorators
You can combine multiple decorators to compose complex behavior:

```python
@app.post("/submit")
@log_request()
@add_custom_header("X-Processed-By", "FastAPI")
@handle_errors()
def submit_data(data: DataModel):
    ...

```

## Dependency injection with parameters
You can pass parameters to your dependencies through closures:

```python
from fastapi_decorators import depends
from fastapi import Depends, HTTPException

def verify_role(required_role: str):
    def dependency(current_user: User = Depends(get_current_user)):
        if current_user.role != required_role:
            raise HTTPException(status_code=403, detail="Forbidden")
    return dependency

@app.get("/admin-area")
@depends(Depends(verify_role("admin")))
def admin_area():
    ...

```

# Credits
Inspired by solutions suggested by [@gocreating](https://github.com/gocreating) and [@dmontagu](https://github.com/dmontagu).

            

Raw data

            {
    "_id": null,
    "home_page": "https://github.com/Minibrams/fastapi-decorators",
    "name": "fastapi-decorators",
    "maintainer": null,
    "docs_url": null,
    "requires_python": ">=3",
    "maintainer_email": null,
    "keywords": "fastapi, decorators, middleware, dependency, dependencies",
    "author": "Anders Brams",
    "author_email": "anders@brams.dk",
    "download_url": "https://files.pythonhosted.org/packages/3c/e6/8986e024d1d22c608c0c077f3b4d8df87d5e8666bf00ed701173315d81ad/fastapi_decorators-1.0.4.tar.gz",
    "platform": null,
    "description": "# FastAPI decorators\n![CI](https://github.com/Minibrams/fastapi-decorators/actions/workflows/ci.yml/badge.svg)\n[![PyPI](https://img.shields.io/pypi/v/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)\n[![Python Versions](https://img.shields.io/pypi/pyversions/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)\n[![Downloads](https://img.shields.io/pypi/dm/fastapi-decorators.svg)](https://pypi.org/project/fastapi-decorators/)\n\nCreate decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.\n\n# Installation\n```bash\npip install fastapi-decorators\n```\n\n# TL;DR\nThe library supplies the `depends()` decorator function which effectively allows you to add argument dependencies to your FastAPI endpoints.\n\nFor example, the following three endpoints have the same signature:\n```python\n# Using normal dependencies\n@app.get(\"/items/{item_id}\")\ndef read_item(item_id: int, _ = Depends(get_current_user)):\n    ...\n\n# Using add_dependency directly\n@app.get(\"/items/{item_id}\")\n@depends(Depends(get_current_user))\ndef read_item(item_id: int):\n    ...\n\n# Using a custom decorator\ndef authorize():\n    def dependency(user = Depends(get_current_user)):\n        return user\n    return depends(Depends(dependency))\n\n@app.get(\"/items/{item_id}\")\n@authorize()\ndef read_item(item_id: int):\n    ...\n```\n\n# Usage examples\n\n- [Using `depends()` directly](#using-depends-directly)\n- [Logging decorator](#logging-decorator)\n- [Authorization decorator](#authorization-decorator)\n- [Custom Response Header decorator](#custom-response-header-decorator)\n- [Rate Limiting decorator](#rate-limiting-decorator)\n- [Caching decorator](#caching-decorator)\n- [Error Handling decorator](#error-handling-decorator)\n- [Combining Multiple decorators](#combining-multiple-decorators)\n- [Dependency injection with parameters](#dependency-injection-with-parameters)\n\n## Using `depends()` directly\nIf you prefer, you can use depends directly without creating a custom decorator:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Depends, Header\n\nasync def verify_api_key(x_api_key: str = Header(...)):\n    if x_api_key != \"expected-api-key\":\n        raise HTTPException(status_code=403, detail=\"Forbidden\")\n\n@app.get(\"/secure-data\")\n@depends(Depends(verify_api_key))\ndef get_secure_data():\n    ...\n\n```\n\n## Logging decorator\nAdd a decorator to log incoming requests:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Request, Depends\n\ndef log_request():\n    def dependency(request: Request):\n        print(f\"Received request: {request.method} {request.url}\")\n    return depends(Depends(dependency))\n\n@app.get(\"/items/{item_id}\")\n@log_request()\ndef read_item(item_id: int):\n    ...\n\n```\n\n## Authorization decorator\nCreate a simple decorator that rejects unauthorized requests:\n> The API docs will reflect the authentication requirement for this endpoint\n> because of the added OAuth2 dependency.\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Depends, HTTPException, Header\nfrom fastapi.security import OAuth2PasswordBearer\nfrom sqlalchemy.orm import Session\n\noauth2_scheme = OAuth2PasswordBearer(tokenUrl=\"token\")\n\ndef authorize(*required_scopes: str):\n    def dependency(token: str = Depends(oauth2_scheme)):\n        # Your auth logic here\n        if not token:\n            raise HTTPException(status_code=401, detail=\"Unauthorized\")\n\n        # Check scopes and permissions\n        pass\n    return depends(Depends(dependency))\n\n\n@app.put(\"/users/{user_id}\")\n@authorize(\"users:write\")\ndef update_user(*, user_id: int, user_update: UserUpdate):\n    ...\n\n```\n\n## Custom Response Header decorator\nCreate a decorator to add custom headers to responses:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Response, Depends\n\ndef add_custom_header(name: str, value: str):\n    def dependency(response: Response):\n        response.headers[name] = value\n    return depends(Depends(dependency))\n\n@app.get(\"/data\")\n@add_custom_header(\"X-Custom-Header\", \"MyValue\")\ndef get_data():\n    ...\n\n```\n\n## Rate Limiting decorator\nAdd rate limiting to your endpoints:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Depends, HTTPException, Request\nfrom time import time\n\nrate_limit_store = {}\n\ndef rate_limit(max_calls: int, period: int):\n    def dependency(ip_address: str = Depends(get_ip_address)):\n        # Simple rate limiting logic\n        now = time()\n        calls, last_reset = rate_limit_store.get(ip_address, (0, now))\n        if now - last_reset > period:\n            # Reset rate limit\n            calls = 0\n            last_reset = now\n        if calls >= max_calls:\n            raise HTTPException(status_code=429, detail=\"Too Many Requests\")\n        calls += 1\n        rate_limit_store[ip_address] = (calls, last_reset)\n    return depends(Depends(dependency))\n\ndef get_ip_address(request: Request):\n    return request.client.host\n\n@app.get(\"/limited-endpoint\")\n@rate_limit(max_calls=5, period=60)\ndef limited_endpoint():\n    ...\n\n```\n\n## Caching decorator\nAdd caching to your endpoints:\n\n```python\ndef get_cache() -> dict:\n    return {}  # Use a real cache like Redis or Memcached\n\ndef cache_response(max_age: int = 5):\n    def decorator(func):\n\n        # Wrap the endpoint after adding the get_cache dependency\n        @depends(cache=Depends(get_cache))\n        @wraps(func)\n        def wrapper(*args, cache: dict, **kwargs):\n            key = func.__name__\n\n            if key in cache:\n                timestamp, data = cache[key]\n                if time() - timestamp < max_age:\n                    # Cache hit\n                    return data\n\n            # Cache miss - call the endpoint as usual\n            result = func(*args, **kwargs)\n\n            # Store the result in the cache\n            cache[key] = time(), result\n            return result\n        \n        return wrapper\n    return decorator\n\n@app.get(\"/cached-data\")\n@cache_response(max_age=10)\ndef get_cached_data():\n    ...\n```\n\n## Error Handling decorator\nCreate a decorator to handle exceptions and return custom responses:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Depends, Response\n\ndef get_crash_log_storage() -> list:\n    return []  # Use a real storage like a database\n\ndef handle_errors():\n    def decorator(func):\n\n        # Wrap the endpoint after adding the crash_logs dependency\n        @depends(crash_logs = Depends(get_crash_log_storage))\n        @wraps(func)\n        def wrapper(*args, crash_logs: list, **kwargs):\n            try:\n                return func(*args, **kwargs)\n            except Exception as e:\n                # Log the error and return a custom response\n                crash_logs.append({ 'error': str(e), 'function': func.__name__ })\n                return JSONResponse(status_code=500, content={ \"detail\": str(e) })\n            \n        return wrapper\n    return decorator\n\n@app.get(\"/may-fail\")\n@handle_errors()\ndef may_fail_operation():\n    ...\n\n```\n\n## Combining Multiple decorators\nYou can combine multiple decorators to compose complex behavior:\n\n```python\n@app.post(\"/submit\")\n@log_request()\n@add_custom_header(\"X-Processed-By\", \"FastAPI\")\n@handle_errors()\ndef submit_data(data: DataModel):\n    ...\n\n```\n\n## Dependency injection with parameters\nYou can pass parameters to your dependencies through closures:\n\n```python\nfrom fastapi_decorators import depends\nfrom fastapi import Depends, HTTPException\n\ndef verify_role(required_role: str):\n    def dependency(current_user: User = Depends(get_current_user)):\n        if current_user.role != required_role:\n            raise HTTPException(status_code=403, detail=\"Forbidden\")\n    return dependency\n\n@app.get(\"/admin-area\")\n@depends(Depends(verify_role(\"admin\")))\ndef admin_area():\n    ...\n\n```\n\n# Credits\nInspired by solutions suggested by [@gocreating](https://github.com/gocreating) and [@dmontagu](https://github.com/dmontagu).\n",
    "bugtrack_url": null,
    "license": null,
    "summary": "Create decorators that leverage FastAPI's `Depends()` and built-in dependencies, enabling you to inject dependencies directly into your decorators.",
    "version": "1.0.4",
    "project_urls": {
        "Homepage": "https://github.com/Minibrams/fastapi-decorators"
    },
    "split_keywords": [
        "fastapi",
        " decorators",
        " middleware",
        " dependency",
        " dependencies"
    ],
    "urls": [
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "c9f314be87d2f13e50d24f21e1e96e39230342672941b344e30cae8cfb5f940a",
                "md5": "db622769078be4418ed97dc1e2eae2a0",
                "sha256": "335a1f13f5d64e87c4e11bb0f6eead1c0aa119d047ffab8a63ed35fc86530f50"
            },
            "downloads": -1,
            "filename": "fastapi_decorators-1.0.4-py3-none-any.whl",
            "has_sig": false,
            "md5_digest": "db622769078be4418ed97dc1e2eae2a0",
            "packagetype": "bdist_wheel",
            "python_version": "py3",
            "requires_python": ">=3",
            "size": 6326,
            "upload_time": "2024-10-03T00:36:02",
            "upload_time_iso_8601": "2024-10-03T00:36:02.168329Z",
            "url": "https://files.pythonhosted.org/packages/c9/f3/14be87d2f13e50d24f21e1e96e39230342672941b344e30cae8cfb5f940a/fastapi_decorators-1.0.4-py3-none-any.whl",
            "yanked": false,
            "yanked_reason": null
        },
        {
            "comment_text": "",
            "digests": {
                "blake2b_256": "3ce68986e024d1d22c608c0c077f3b4d8df87d5e8666bf00ed701173315d81ad",
                "md5": "69c252c1c9f81184988c9a1935015550",
                "sha256": "22c4d1e54d157c29de4a14a963e550d42c42867600e62d7b7bfb7bd33a9a9a0b"
            },
            "downloads": -1,
            "filename": "fastapi_decorators-1.0.4.tar.gz",
            "has_sig": false,
            "md5_digest": "69c252c1c9f81184988c9a1935015550",
            "packagetype": "sdist",
            "python_version": "source",
            "requires_python": ">=3",
            "size": 7158,
            "upload_time": "2024-10-03T00:36:03",
            "upload_time_iso_8601": "2024-10-03T00:36:03.989676Z",
            "url": "https://files.pythonhosted.org/packages/3c/e6/8986e024d1d22c608c0c077f3b4d8df87d5e8666bf00ed701173315d81ad/fastapi_decorators-1.0.4.tar.gz",
            "yanked": false,
            "yanked_reason": null
        }
    ],
    "upload_time": "2024-10-03 00:36:03",
    "github": true,
    "gitlab": false,
    "bitbucket": false,
    "codeberg": false,
    "github_user": "Minibrams",
    "github_project": "fastapi-decorators",
    "travis_ci": false,
    "coveralls": false,
    "github_actions": true,
    "lcname": "fastapi-decorators"
}
        
Elapsed time: 0.37888s