105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""
|
|
Utility functions for API event handling.
|
|
"""
|
|
|
|
from typing import TypeVar, Callable, Dict, Any
|
|
from functools import wraps
|
|
from fastapi import Request
|
|
|
|
R = TypeVar('R')
|
|
|
|
class BaseEndpointHandler:
|
|
"""Base class for handling endpoint execution with context."""
|
|
|
|
def __init__(self, func: Callable, url_of_endpoint: str):
|
|
self.func = func
|
|
self.url_of_endpoint = url_of_endpoint
|
|
self.function_code = url_of_endpoint # Set initial function_code
|
|
self._context = {
|
|
'url_of_endpoint': url_of_endpoint,
|
|
'function_code': url_of_endpoint, # Initialize with URL
|
|
}
|
|
|
|
@property
|
|
def context(self) -> dict:
|
|
"""Get the endpoint context."""
|
|
return self._context
|
|
|
|
def update_context(self, **kwargs):
|
|
"""Update the endpoint context with new values."""
|
|
self._context.update(kwargs)
|
|
# Update function_code property if it's in the context
|
|
if 'function_code' in kwargs:
|
|
self.function_code = kwargs['function_code']
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
return self.func(*args, **kwargs)
|
|
|
|
|
|
class TokenEventHandler(BaseEndpointHandler):
|
|
"""Handler for endpoints that require token and event tracking."""
|
|
|
|
def __init__(self, func: Callable, url_of_endpoint: str):
|
|
super().__init__(func, url_of_endpoint)
|
|
self.update_context(
|
|
token_dict={
|
|
'user_id': '1234567890',
|
|
'username': 'test_user',
|
|
'email': 'asda@email.com',
|
|
}
|
|
)
|
|
|
|
|
|
class AuthHandler(BaseEndpointHandler):
|
|
"""Handler for endpoints that require only authentication."""
|
|
|
|
def __init__(self, func: Callable, url_of_endpoint: str):
|
|
super().__init__(func, url_of_endpoint)
|
|
self.update_context(
|
|
auth_level="user",
|
|
permissions=["read", "write"]
|
|
)
|
|
|
|
|
|
def with_token_event(func: Callable[..., Dict[str, Any]]) -> Callable[..., Dict[str, Any]]:
|
|
"""Decorator for endpoints with token and event requirements."""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs) -> Dict[str, Any]:
|
|
# Create handler with context
|
|
handler = TokenEventHandler(
|
|
func=func,
|
|
url_of_endpoint=func.url_of_endpoint
|
|
)
|
|
|
|
# Update event-specific context
|
|
handler.update_context(
|
|
function_code=f"7192c2aa-5352-4e36-98b3-dafb7d036a3d" # Keep function_code as URL
|
|
)
|
|
|
|
# Make context available to the function
|
|
func.handler = handler
|
|
|
|
# Call the original function
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
def auth_required(func: Callable[..., Dict[str, Any]]) -> Callable[..., Dict[str, Any]]:
|
|
"""Decorator for endpoints with only auth requirements."""
|
|
@wraps(func)
|
|
def wrapper(*args, **kwargs) -> Dict[str, Any]:
|
|
# Create handler with context
|
|
handler = AuthHandler(
|
|
func=func,
|
|
url_of_endpoint=func.url_of_endpoint
|
|
)
|
|
|
|
# Make context available to the function
|
|
func.handler = handler
|
|
|
|
# Call the original function
|
|
return func(*args, **kwargs)
|
|
|
|
return wrapper
|