#23 - first draft of CleverBRAG facade with endpoints discovery, suggested type (de)serialization and sample invocation
This commit is contained in:
Stanislav Hejny
@@ -0,0 +1,389 @@
|
||||
"""
|
||||
This module provides the CleverBRAG AMQP Adapter for handling API requests via AMQP.
|
||||
The code IS MEANT to be part of CleverBRAG codebase, and has direct dependencies on the CleverBRAG codebase.
|
||||
It is NOT MEANT to be used as a standalone module. It WILL NOT run outside the CleverBRAG context.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import re
|
||||
from collections import defaultdict
|
||||
from threading import Thread
|
||||
from typing import List, Dict, Any
|
||||
from fastapi.routing import APIRoute
|
||||
from fastapi.security import OAuth2PasswordRequestForm
|
||||
from fastapi.security.http import HTTPAuthorizationCredentials
|
||||
from pydantic import BaseModel
|
||||
from uuid import UUID
|
||||
from starlette.datastructures import UploadFile, Headers
|
||||
import pathlib
|
||||
|
||||
# CleverBrag specific imports
|
||||
import core
|
||||
from core import R2RConfig, R2RBuilder, R2RProviderFactory
|
||||
from sdk.async_client import create_r2r_app
|
||||
from shared.abstractions import R2RSerializable, User
|
||||
|
||||
# AMQP specific imports
|
||||
from amqp.adapter.data_response_factory import AMQResponseFactory
|
||||
from amqp.adapter.data_parser import AMQDataParser
|
||||
from amqp.adapter.cleverthis_service_adapter import CleverThisServiceAdapter, AMQMessage
|
||||
from amqp.config.amq_configuration import AMQConfiguration
|
||||
from amqp.model.model import AMQResponse
|
||||
from amqp.service.amq_service import AMQService
|
||||
|
||||
|
||||
def clone_and_adapt_route(original_route: APIRoute) -> APIRoute:
|
||||
"""
|
||||
Creates a clone of an APIRoute. It checks if there's preferred endpoint with json serialization.
|
||||
It also adjusts the path regex to ensure it matches the path in different API versions.
|
||||
"""
|
||||
|
||||
# Need to adapt also the path regex to ensure it matches the path in different API versions
|
||||
# Ensures the leading '^' is followed by any match and there's no '/' before final '$'
|
||||
_new_regex_string = (
|
||||
"^.*" + original_route.path_regex.pattern.lstrip("^").rstrip("/$") + "$"
|
||||
)
|
||||
_new_route: APIRoute = APIRoute(
|
||||
path=original_route.path,
|
||||
endpoint=original_route.endpoint,
|
||||
response_model=original_route.response_model,
|
||||
status_code=original_route.status_code,
|
||||
tags=original_route.tags.copy() if original_route.tags else None,
|
||||
dependencies=(
|
||||
original_route.dependencies.copy() if original_route.dependencies else None
|
||||
),
|
||||
summary=original_route.summary,
|
||||
description=original_route.description,
|
||||
response_description=original_route.response_description,
|
||||
responses=original_route.responses.copy() if original_route.responses else None,
|
||||
deprecated=original_route.deprecated,
|
||||
name=original_route.name,
|
||||
methods=original_route.methods.copy() if original_route.methods else None,
|
||||
operation_id=original_route.operation_id,
|
||||
response_model_include=original_route.response_model_include,
|
||||
response_model_exclude=original_route.response_model_exclude,
|
||||
response_model_by_alias=original_route.response_model_by_alias,
|
||||
response_model_exclude_unset=original_route.response_model_exclude_unset,
|
||||
response_model_exclude_defaults=original_route.response_model_exclude_defaults,
|
||||
response_model_exclude_none=original_route.response_model_exclude_none,
|
||||
include_in_schema=original_route.include_in_schema,
|
||||
response_class=original_route.response_class,
|
||||
dependency_overrides_provider=original_route.dependency_overrides_provider,
|
||||
callbacks=original_route.callbacks.copy() if original_route.callbacks else None,
|
||||
openapi_extra=(
|
||||
original_route.openapi_extra.copy()
|
||||
if original_route.openapi_extra
|
||||
else None
|
||||
),
|
||||
generate_unique_id_function=original_route.generate_unique_id_function,
|
||||
)
|
||||
_new_route.path_regex = re.compile(_new_regex_string)
|
||||
return _new_route
|
||||
|
||||
|
||||
def group_routes_by_method(api_routes: List[APIRoute]) -> Dict[str, List[APIRoute]]:
|
||||
"""
|
||||
Groups API routes by their HTTP method. This is important to ensure that the URLPath is matched
|
||||
correctly, as different methods may share the same URLPath (e.g. GET, POST PUT, DELETE).
|
||||
"""
|
||||
_method_to_routes = defaultdict(
|
||||
list
|
||||
) # Automatically initializes new lists for new keys
|
||||
for _route in api_routes:
|
||||
if (
|
||||
isinstance(_route, APIRoute) and _route.methods
|
||||
): # Ensure methods exist (should always be true for valid routes)
|
||||
for method in _route.methods:
|
||||
_method_to_routes[method].append(clone_and_adapt_route(_route))
|
||||
|
||||
return dict(_method_to_routes)
|
||||
|
||||
|
||||
class BRAGArgument:
|
||||
"""
|
||||
This class is used to represent a BRAG API input argument. It is uses FastAPI annotation to define the arguments
|
||||
(the argument type) that are passed as input arguments to the BRAG endpoint.
|
||||
"""
|
||||
|
||||
def __init__(self, arg: Any):
|
||||
self.arg = arg
|
||||
self.module = arg.__module__
|
||||
self.cls = arg.__name__ if hasattr(arg, "__name__") else str(arg)
|
||||
self.is_r2r_serializable = issubclass(arg, R2RSerializable)
|
||||
self.is_builtin = self.module == "builtins"
|
||||
self.is_uuid = self.cls == "UUID"
|
||||
self.is_optional = self.cls == "Optional"
|
||||
self.is_list = self.cls == "list"
|
||||
self.is_dict = self.cls == "dict"
|
||||
self.is_base_model = issubclass(arg, BaseModel)
|
||||
self.is_nested = self.is_list or self.is_optional or self.is_dict
|
||||
if self.is_nested and hasattr(arg, "__args__"):
|
||||
self.nested = BRAGArgument(getattr(arg, "__args__")[0])
|
||||
if self.is_dict:
|
||||
self.nested = (self.nested, BRAGArgument(getattr(arg, "__args__")[1]))
|
||||
else:
|
||||
self.nested = None
|
||||
self.is_nested = False
|
||||
|
||||
def __repr__(self):
|
||||
return f"{self.module}.{self.cls}" + (
|
||||
f"[{self.nested}]" if self.is_nested else ""
|
||||
)
|
||||
|
||||
|
||||
def convert_to_argument(arg: BRAGArgument, value: str) -> Any:
|
||||
"""
|
||||
Converts the argument to a string representation.
|
||||
"""
|
||||
try:
|
||||
if arg.cls == "str":
|
||||
return value
|
||||
elif arg.cls == "bool":
|
||||
return str(value).lower() == "true"
|
||||
elif arg.cls == "float":
|
||||
return float(value)
|
||||
elif arg.cls == "int":
|
||||
return int(value)
|
||||
elif arg.is_uuid:
|
||||
return UUID(value)
|
||||
elif arg.is_optional:
|
||||
return convert_to_argument(arg.nested, value) if arg.nested else None
|
||||
elif arg.is_list:
|
||||
return (
|
||||
[convert_to_argument(arg.nested, v.strip()) for v in value.split(",")]
|
||||
if arg.nested
|
||||
else value.split()
|
||||
)
|
||||
elif arg.is_dict:
|
||||
return (
|
||||
{
|
||||
convert_to_argument(arg.nested[0], k.strip()): convert_to_argument(
|
||||
arg.nested[1], v.strip()
|
||||
)
|
||||
for k, v in (item.split(":") for item in value.split(","))
|
||||
}
|
||||
if arg.nested
|
||||
else dict(item.split(":") for item in value.split(","))
|
||||
)
|
||||
elif arg.is_base_model:
|
||||
return arg.arg.parse_raw(value)
|
||||
else:
|
||||
print(f"Unsupported argument type: {type(arg)}")
|
||||
except Exception as e:
|
||||
print(f"Error converting argument {arg}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
class CleverBragAmqpAdapter(CleverThisServiceAdapter):
|
||||
"""
|
||||
CleverBRAG AMQP Adapter for CleverBRAG API. Implements the interface to CleverBRAG API.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
amq_configuration: AMQConfiguration,
|
||||
routes: List[APIRoute],
|
||||
auth_provider: core.base.providers.auth.AuthProvider,
|
||||
):
|
||||
super().__init__(amq_configuration, None)
|
||||
self.routes = group_routes_by_method(api_routes=routes)
|
||||
self.auth_provider = auth_provider
|
||||
_amq_service: AMQService = AMQService(amq_configuration, self)
|
||||
thread = Thread(target=_amq_service.run)
|
||||
thread.start()
|
||||
# thread.join()
|
||||
|
||||
async def get_current_user(self, token: str) -> User:
|
||||
"""
|
||||
Overrides the default get_current_user method to use the token from the AMQP message.
|
||||
:param token: JWT token from the AMQP message
|
||||
:return: User object
|
||||
"""
|
||||
return await self.auth_provider.auth_wrapper(
|
||||
HTTPAuthorizationCredentials(f"Bearer {token}")
|
||||
)
|
||||
|
||||
# ==================================================================================================
|
||||
# ===================== C l e v e r S w a r m A P I m e t h o d s ============================
|
||||
# ==================================================================================================
|
||||
|
||||
async def get(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the GET requests for the CleverBRAG API.
|
||||
:param message: message from the AMQP that contains the GET request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
return await self.handle_no_body_payload(message, auth_user, "GET")
|
||||
|
||||
async def put(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the PUT requests for the CleverBRAG API.
|
||||
:param message: message from the AMQP that contains the PUT request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
return await self.handle_body_payload(message, auth_user, "PUT")
|
||||
|
||||
async def post(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the POST requests for the CleverBRAG API.
|
||||
:param message: message from the AMQP that contains the POST request, including body and headers
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
return await self.handle_body_payload(message, auth_user, "POST")
|
||||
|
||||
async def head(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the HEAD requests for the CleverBRAG API.
|
||||
"""
|
||||
return await self.handle_no_body_payload(message, auth_user, "HEAD")
|
||||
|
||||
async def delete(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the DELETE requests for the CleverBRAG API.
|
||||
"""
|
||||
return await self.handle_no_body_payload(message, auth_user, "DELETE")
|
||||
|
||||
async def options(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the OPTIONS requests for the CleverBRAG API.
|
||||
"""
|
||||
return await self.handle_no_body_payload(message, auth_user, "OPTIONS")
|
||||
|
||||
async def patch(self, message: AMQMessage, auth_user: User) -> AMQResponse:
|
||||
"""
|
||||
Handles the PATCH requests for the CleverBRAG API.
|
||||
"""
|
||||
return await self.handle_no_body_payload(message, auth_user, "PATCH")
|
||||
|
||||
async def handle_no_body_payload(
|
||||
self, message: AMQMessage, auth_user: User, method: str
|
||||
) -> AMQResponse:
|
||||
"""
|
||||
Handles the requests that do not have a body payload.
|
||||
:param message: message from the AMQP that contains the request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
_routes = self.routes.get(method, [])
|
||||
_path, _args = AMQDataParser.parse_get_url(message.path)
|
||||
for _route in _routes:
|
||||
_match = _route.path_regex.match(_path)
|
||||
if _match:
|
||||
_args.update(_match.groupdict())
|
||||
_args["authenticated_user"] = auth_user
|
||||
# If the path matches, call the corresponding function
|
||||
return await CleverThisServiceAdapter.call_cleverthis_api(
|
||||
message,
|
||||
_args,
|
||||
api_method=_route.endpoint,
|
||||
success_code=_route.status_code,
|
||||
loop=self.loop,
|
||||
)
|
||||
|
||||
return AMQResponseFactory.of(
|
||||
message.id,
|
||||
404,
|
||||
"text/plain",
|
||||
str(f"Destination location [{message.path}] does not exists").encode(
|
||||
"utf-8"
|
||||
),
|
||||
message.trace_info,
|
||||
)
|
||||
|
||||
async def handle_body_payload(
|
||||
self, message: AMQMessage, auth_user: User, method: str
|
||||
) -> AMQResponse:
|
||||
"""
|
||||
Handles the requests that have a body payload, including Multipart data.
|
||||
:param message: message from the AMQP that contains the request
|
||||
:param auth_user: user authenticated by the API Gateway, as UserSchem
|
||||
:param method: HTTP method (POST, PUT, etc.)
|
||||
:return: AMQResponse to be sent back to the API Gateway via AMQP
|
||||
"""
|
||||
_form_data = AMQDataParser.parse_request_body(message)
|
||||
_routes = self.routes.get(method, [])
|
||||
_path, _query_args = AMQDataParser.parse_get_url(message.path)
|
||||
for _route in _routes:
|
||||
_match = _route.path_regex.match(_path)
|
||||
if _match:
|
||||
# If the path matches, call the corresponding function
|
||||
_form_data.update(_match.groupdict())
|
||||
_form_data.update(_query_args)
|
||||
for _body_param in _route.dependant.body_params:
|
||||
if _body_param.name in _form_data:
|
||||
_file_data = _form_data[_body_param.name]
|
||||
actual_filepath = message.extra_data[_body_param.name]
|
||||
_form_data[_body_param.name] = (
|
||||
_body_param.type_.__name__ == "UploadFile"
|
||||
and UploadFile(
|
||||
file=pathlib.Path(actual_filepath).open("rb"),
|
||||
size=_file_data["size"],
|
||||
filename=_file_data["filename"],
|
||||
headers=Headers(
|
||||
{"Content-Type": _file_data["mediaType"]}
|
||||
),
|
||||
)
|
||||
or _form_data[_body_param.name]
|
||||
)
|
||||
|
||||
_verified_args = {}
|
||||
_required_args = getattr(_route.endpoint, "__annotations__", {})
|
||||
for arg in _required_args:
|
||||
if arg in _form_data:
|
||||
_verified_args[arg] = _form_data[arg]
|
||||
else:
|
||||
# needed for /login input
|
||||
if arg == "authenticated_user":
|
||||
_verified_args["authenticated_user"] = auth_user
|
||||
if _required_args[arg].__name__ == "OAuth2PasswordRequestForm":
|
||||
try:
|
||||
_verified_args[arg] = OAuth2PasswordRequestForm(
|
||||
**_form_data
|
||||
)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
return await CleverThisServiceAdapter.call_cleverthis_api(
|
||||
message,
|
||||
_verified_args,
|
||||
api_method=_route.endpoint,
|
||||
success_code=_route.status_code if _route.status_code else 200,
|
||||
loop=self.loop,
|
||||
)
|
||||
|
||||
return AMQResponseFactory.of(
|
||||
message.id,
|
||||
404,
|
||||
"text/plain",
|
||||
str(f"Destination location [{message.path}] does not exists").encode(
|
||||
"utf-8"
|
||||
),
|
||||
message.trace_info,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
# Example usage
|
||||
r2rapp = asyncio.run(create_r2r_app())
|
||||
|
||||
adapter: CleverBragAmqpAdapter = CleverBragAmqpAdapter(
|
||||
AMQConfiguration(""), r2rapp.app.routes, r2rapp.providers.auth
|
||||
)
|
||||
|
||||
for method, routes in adapter.routes.items():
|
||||
print(f"Method: {method}")
|
||||
for _route in routes:
|
||||
print(
|
||||
f" Route: {_route.path} -> {_route.endpoint.__module__}.{_route.endpoint.__name__}"
|
||||
)
|
||||
_args = getattr(_route.endpoint, "__annotations__", {})
|
||||
_verified_args = []
|
||||
for arg in _args:
|
||||
_arg = BRAGArgument(_args[arg])
|
||||
_verified_args.append(_arg)
|
||||
print(f" Arguments: {', '.join(map(str, _verified_args))}")
|
||||
print("---------------------------")
|
||||
Reference in New Issue
Block a user