Decorator which ensures we are routing our db operations through a backend if configured, or directly if not.
route(route_endpoint, method='POST', send_file=False)
Handle routing to backend or direct database operations based on environment configuration via decorator.
Parameters:
| Name |
Type |
Description |
Default |
route_endpoint
|
str
|
The backend endpoint to route to if ROUTE_TO_BACKEND is True.
|
required
|
method
|
request_methods
|
If routing through backend, the request method. Default: POST.
|
'POST'
|
send_file
|
bool
|
If True sends a file to backend. Otherwise, json. Default: False.
|
False
|
Returns:
| Type |
Description |
|
|
Decorator function that handles routing logic.
|
Source code in opensampl/load/routing.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114 | def route(route_endpoint: str, method: request_methods = "POST", send_file: bool = False):
"""
Handle routing to backend or direct database operations based on environment configuration via decorator.
Args:
route_endpoint: The backend endpoint to route to if ROUTE_TO_BACKEND is True.
method: If routing through backend, the request method. Default: POST.
send_file: If True sends a file to backend. Otherwise, json. Default: False.
Returns:
Decorator function that handles routing logic.
"""
def decorator(func: Callable) -> Callable:
"""
Wrap the function with routing logic.
Args:
func: The function to be decorated.
Returns:
Wrapped function with routing capabilities.
"""
@wraps(func)
def wrapper(*args: list, **kwargs: dict) -> Optional[Callable]:
"""
Handle the actual routing logic.
Args:
*args: Positional arguments passed to the wrapped function.
**kwargs: Keyword arguments passed to the wrapped function.
Returns:
Result from either backend request or direct function call.
Raises:
requests.exceptions.RequestException: If backend request fails.
"""
session = kwargs.pop("session", None)
config = BaseConfig()
config.check_routing_dependencies()
logger.debug(f"{config.ROUTE_TO_BACKEND=}")
# Config Validation deals with making sure we have a backend url if going through backend and
# a database url if we are doing db operations directly
if config.ROUTE_TO_BACKEND:
headers = {
"access-key": config.API_KEY,
}
pyld = func(*args, **kwargs, _config=config)
if send_file:
request_params = pyld
logger.debug(f"data={pyld.get('data')}")
logger.debug(f"filesize in bytes={len(pyld.get('files').get('file')[1])}")
else:
request_params = {
"json": pyld,
}
headers.update({"Content-Type": "application/json"})
logger.debug(f"headers={json.dumps(headers, indent=4)}")
logger.debug(f"json={json.dumps(pyld, indent=4)}")
# Extract data from the function
try:
logger.debug(f"method={method} type={type(method)}")
logger.debug(f"request url={config.BACKEND_URL}/{route_endpoint}")
response = requests.request(
method=str(method),
url=f"{config.BACKEND_URL}/{route_endpoint}",
headers=headers,
**request_params,
timeout=300,
verify=not config.INSECURE_REQUESTS,
)
logger.debug(f"{response.request.method=}, {response.status_code=}, {response.url=}")
response.raise_for_status()
logger.debug(f"Response: {response.json()}")
except requests.exceptions.RequestException as e:
logger.debug(f"Error making request to backend: {e}")
raise
else:
if not session:
session = sessionmaker(create_engine(config.DATABASE_URL))() # ty: ignore[no-matching-overload]
return func(*args, **kwargs, session=session, _config=config)
return wrapper
return decorator
|