What Are Function Decorators?

A decorator is a callable that takes a function as an input and returns a new function. This new function typically extends or modifies the behavior of the original function.

The basic syntax for using a decorator is:

1@decorator_function
2def target_function():
3    pass

This is equivalent to:

1def target_function():
2    pass
3target_function = decorator_function(target_function)

In this example, uppercase_decorator is a function that takes another function as an argument. Read: Python Functions and Built-In Functions It defines an inner function wrapper that calls the original function, converts its result to uppercase, and returns it.

 1def uppercase_decorator(func):
 2    def wrapper():
 3        result = func()
 4        return result.upper()
 5    return wrapper
 6
 7@uppercase_decorator
 8def greet():
 9    return "hello, world!"
10
11print(greet())  
12# HELLO, WORLD!

Function Decorators with Arguments

 1def repeat(times):
 2    def decorator(func):
 3        def wrapper(*args, **kwargs):
 4            for _ in range(times):
 5                result = func(*args, **kwargs)
 6            return result
 7        return wrapper
 8    return decorator
 9
10@repeat(3)
11def say_hello(name):
12    print(f"Hello, {name}!")
13
14say_hello("Alice")
15# Hello, Alice!
16# Hello, Alice!
17# Hello, Alice!

Real-World Use Cases of Synchronous Decorators

Decorators are not just a syntactic nicety. They have several practical applications:

  1. Logging: You can use decorators to add logging to functions without cluttering the function’s code.
  2. Timing: Measure the execution time of functions.
  3. Caching: Implement memoization to cache expensive function calls.
  4. Authentication: Check if a user is authenticated before allowing access to certain functions.
  5. Debugging: Print out arguments of the given function.

Read: Mastering Asynchronous Decorators in Python

Function Async Decorators

Async decorators are used to decorate coroutines (async functions) in Python. They’re particularly useful when working with asynchronous code, such as in web applications or when dealing with I/O-bound operations.

 1import asyncio
 2import time
 3from functools import wraps
 4
 5def async_timing_decorator(func):
 6    @wraps(func)
 7    async def wrapper(*args, **kwargs):
 8        start_time = time.time()
 9        result = await func(*args, **kwargs)
10        end_time = time.time()
11        print(f"{func.__name__} took {end_time - start_time:.2f} seconds to execute.")
12        return result
13    return wrapper
14
15@async_timing_decorator
16async def async_slow_function():
17    await asyncio.sleep(2)
18    print("Async function execution complete.")
19
20async def main():
21    await async_slow_function()
22
23asyncio.run(main())

Let’s break down the function async_timing_decorator

  1. We import asyncio for asynchronous operations, time for timing, and wraps from functools to preserve the metadata of the decorated function.
  2. The async_timing_decorator is defined similarly to the synchronous version, but with some key differences:
    • The inner wrapper function is defined with async def, making it a coroutine.
    • Inside wrapper, we use await to call the decorated function.
  3. We use @wraps(func) to preserve the metadata of the original function. This is a good practice for all decorators, but especially important for async functions where introspection is often used.
  4. The async_slow_function is defined with async def and uses await asyncio.sleep(2) to simulate an asynchronous operation.
  5. We define a main() coroutine to run our async function.
  6. Finally, we use asyncio.run(main()) to run the async code in the event loop.

A More Realistic Scenario of Asynchronous Decorators

Here’s an example of how you might use this in a more realistic scenario with an async web framework like FastAPI:

 1from fastapi import FastAPI
 2import aiohttp
 3import time
 4from functools import wraps
 5
 6app = FastAPI()
 7
 8def async_timing_decorator(func):
 9    @wraps(func)
10    async def wrapper(*args, **kwargs):
11        start_time = time.time()
12        result = await func(*args, **kwargs)
13        end_time = time.time()
14        print(f"{func.__name__} took {end_time - start_time:.2f} seconds to execute.")
15        return result
16    return wrapper
17
18@async_timing_decorator
19async def fetch_data(url: str):
20    async with aiohttp.ClientSession() as session:
21        async with session.get(url) as response:
22            return await response.json()
23
24@app.get("/data")
25async def get_data():
26    data = await fetch_data("https://api.example.com/data")
27    return {"data": data}

In this example, the async_timing_decorator would help you monitor the performance of your external API calls, which can be crucial for optimizing web application performance.

Stacking Layers of Decorators: Enhancing Functions with Multiple Capabilities

This example demonstrates how async decorators can be used to add various layers of functionality to your FastAPI routes in a clean and reusable way. You can, easily, apply these decorators to any number of routes, keeping your code DRY and maintainable.

 1from fastapi import FastAPI, HTTPException, Depends
 2from fastapi.security import OAuth2PasswordBearer
 3from typing import Callable
 4import time
 5from functools import wraps
 6
 7app = FastAPI()
 8
 9# Simulated database of users
10fake_users_db = {
11    "johndoe": {"username": "johndoe", "full_name": "John Doe", "email": "[email protected]"},
12    "alice": {"username": "alice", "full_name": "Alice Wonderland", "email": "[email protected]"},
13}
14
15oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
16
17# Async decorator for authentication
18def async_auth_required(func: Callable):
19    @wraps(func)
20    async def wrapper(*args, **kwargs):
21        token = await oauth2_scheme(kwargs.get('token'))
22        if token not in fake_users_db:
23            raise HTTPException(status_code=401, detail="Invalid authentication credentials")
24        return await func(*args, **kwargs)
25    return wrapper
26
27# Async decorator for rate limiting
28def async_rate_limit(calls: int, period: int):
29    def decorator(func: Callable):
30        calls_made = {}
31        @wraps(func)
32        async def wrapper(*args, **kwargs):
33            now = time.time()
34            if func.__name__ not in calls_made:
35                calls_made[func.__name__] = []
36            calls_made[func.__name__] = [call for call in calls_made[func.__name__] if call > now - period]
37            if len(calls_made[func.__name__]) >= calls:
38                raise HTTPException(status_code=429, detail="Rate limit exceeded")
39            calls_made[func.__name__].append(now)
40            return await func(*args, **kwargs)
41        return wrapper
42    return decorator
43
44# Async decorator for error handling
45def async_error_handler(func: Callable):
46    @wraps(func)
47    async def wrapper(*args, **kwargs):
48        try:
49            return await func(*args, **kwargs)
50        except Exception as e:
51            raise HTTPException(status_code=500, detail=f"An error occurred: {str(e)}")
52    return wrapper
53
54# Async decorator for timing
55def async_timing_decorator(func: Callable):
56    @wraps(func)
57    async def wrapper(*args, **kwargs):
58        start_time = time.time()
59        result = await func(*args, **kwargs)
60        end_time = time.time()
61        print(f"{func.__name__} took {end_time - start_time:.2f} seconds to execute.")
62        return result
63    return wrapper
64
65@app.get("/users/me")
66@async_auth_required
67@async_rate_limit(calls=5, period=60)
68@async_error_handler
69@async_timing_decorator
70async def read_users_me(token: str = Depends(oauth2_scheme)):
71    return fake_users_db[token]
72
73if __name__ == "__main__":
74   import uvicorn
75   uvicorn.run(app, host="0.0.0.0", port=8000)

Let’s break down this example:

We define several async decorators:

async_auth_required: Checks if the provided token is valid. async_rate_limit: Limits the number of calls to a function within a specific time period. async_error_handler: Catches any exceptions and returns them as HTTP errors. async_timing_decorator: Measures and logs the execution time of a function.

This endpoint /users/me:

  • Returns user information.
  • It’s protected by authentication.
  • Rate-limited to 5 calls per minute.
  • Error handling.
  • The Function is Timed.

The decorators are stacked on top of each other. The order matters: the outermost decorator (bottom in the stack) is applied first, and the innermost (top in the stack) is applied last.

To use this API:

Run the server using uvicorn (as shown in the if __name__ == "__main__": block). You can then make requests to these endpoints. For example:

GET /users/me with a valid token in the header.

1{ 
2   "johndoe": {
3      "username": "johndoe", "full_name": "John Doe", "email": "[email protected]"
4   }
5}

Notes:

  • Use functools.wraps to preserve function metadata, adhering to Python coding standards.
  • Consider the order when stacking multiple decorators for optimal functionality.
  • Simple decorators can modify function output (e.g., converting to uppercase), improving code reusability.
  • Decorators can accept arguments for more flexible behavior, enhancing Python’s dynamic nature.
  • Common use cases include logging, timing, caching, debugging, and authentication.

Read: Python @property decorator