Feat/swagger execute (#25)

* chore: update openapi utils for not depending on FastAPI version

* feat: add rpc execute utils

Author: Kiruha01

app/api/endpoints/swagger_rpc_execute.py

@@ -0,0 +1,61 @@
+"""Execute RPC method endpoint."""
+from json import JSONDecodeError
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse
+from pybotx import Bot
+from pybotx_smartapp_rpc import RPCErrorResponse, SmartApp
+from pybotx_smartapp_rpc.models.request import RPCRequest
+from starlette.status import HTTP_200_OK, HTTP_400_BAD_REQUEST
+
+from app.api.dependencies.bot import bot_dependency
+from app.services.execute_rpc import (
+    RPCAuthConfig,
+    event_factory,
+    expand_config,
+    security,
+)
+from app.smartapp.smartapp import smartapp as smartapp_rpc
+
+router = APIRouter(include_in_schema=False)
+
+
+@router.post("/{method:str}", response_class=JSONResponse)
+async def rpc_execute(
+    method: str,
+    request: Request,
+    credentials: RPCAuthConfig = Depends(security),
+    bot: Bot = bot_dependency,
+) -> JSONResponse:
+    """Execute RPC method."""
+    bot_account, user_info = await expand_config(credentials, bot)
+
+    try:
+        method_payload = await request.json()
+    except JSONDecodeError:
+        method_payload = {}
+
+    event = event_factory(
+        method,
+        method_payload,
+        bot_account,
+        user_info,
+        credentials.sender_udid,
+        credentials.chat_id,
+    )
+
+    smartapp = SmartApp(bot, event.bot.id, event.chat.id, event)
+    rpc_request = RPCRequest(method=method, type="smartapp_rpc", params=method_payload)
+
+    rpc_response = await smartapp_rpc._router.perform_rpc_request(  # noqa: WPS437
+        smartapp, rpc_request
+    )
+    if isinstance(rpc_response, RPCErrorResponse):
+        return JSONResponse(
+            status_code=HTTP_400_BAD_REQUEST,
+            content=rpc_response.jsonable_dict()["errors"],
+        )
+
+    return JSONResponse(
+        status_code=HTTP_200_OK, content=rpc_response.jsonable_dict().get("result")
+    )

app/api/routers.py

@@ -3,8 +3,11 @@
 
 from app.api.endpoints.botx import router as bot_router
 from app.api.endpoints.healthcheck import router as healthcheck_router
+from app.api.endpoints.swagger_rpc_execute import router as swagger_rpc_execute_router
 
 router = APIRouter(include_in_schema=False)
 
 router.include_router(healthcheck_router)
 router.include_router(bot_router)
+
+router.include_router(swagger_rpc_execute_router)

app/main.py

@@ -76,6 +76,7 @@ def get_custom_openapi() -> Dict[str, Any]:  # noqa: WPS430
             version="0.1.0",
             fastapi_routes=application.routes,
             rpc_router=smartapp.router,
+            openapi_version="3.0.2",
         )
 
     application.openapi = get_custom_openapi  # type: ignore

app/services/execute_rpc.py

@@ -0,0 +1,146 @@
+"""RPC execution service."""
+import re
+from typing import Any, Dict, Tuple
+from uuid import UUID, uuid4
+
+from fastapi import HTTPException
+from fastapi.security import APIKeyHeader
+from pybotx import (  # noqa: WPS235
+    Bot,
+    BotAccount,
+    Chat,
+    ChatTypes,
+    SmartAppEvent,
+    UserDevice,
+    UserFromSearch,
+    UserKinds,
+    UserNotFoundError,
+    UserSender,
+)
+from pydantic import BaseModel, ValidationError
+from starlette.requests import Request
+from starlette.status import HTTP_400_BAD_REQUEST, HTTP_403_FORBIDDEN
+
+from app.settings import settings
+
+DOCS = """Установка параметров для выполнение RPC методов.
+
+* `bot_id` - huid бота. Необязательное поле.
+* `sender_huid` - huid пользователя. Необязательное поле.
+* `sender_udid` - udid пользователя. Необязательное поле.
+* `chat_id` - id чата. Необязательное поле.
+
+**Example**: `bot_id=UUID&sender_huid=UUID&sender_udid=UUID&chat_id=UUID`"""
+
+
+class RPCAuthConfig(BaseModel):
+    bot_id: UUID = settings.BOT_CREDENTIALS[0].id
+    sender_huid: UUID = uuid4()
+    sender_udid: UUID = uuid4()
+    chat_id: UUID = uuid4()
+
+
+async def expand_config(
+    config: RPCAuthConfig, bot: Bot
+) -> Tuple[BotAccount, UserFromSearch]:
+    bot_account = [bot for bot in bot.bot_accounts if bot.id == config.bot_id]
+    if not bot_account:
+        raise HTTPException(
+            status_code=HTTP_400_BAD_REQUEST,
+            detail=f"Bot with id {config.bot_id} not found",
+        )
+    try:
+        user_info = await bot.search_user_by_huid(
+            bot_id=bot_account[0].id, huid=config.sender_huid
+        )
+    except UserNotFoundError:
+        user_info = UserFromSearch(
+            huid=config.sender_huid,
+            ad_login=None,
+            ad_domain=None,
+            username="Username",
+            company=None,
+            company_position=None,
+            department=None,
+            emails=[],
+            other_id=None,
+            user_kind=UserKinds.CTS_USER,
+        )
+
+    return bot_account[0], user_info
+
+
+def event_factory(
+    method_name: str,
+    payload: Dict[str, Any],
+    bot_account: BotAccount,
+    user_info: UserFromSearch,
+    user_udid: UUID,
+    chat_id: UUID,
+) -> SmartAppEvent:
+    return SmartAppEvent(
+        bot=BotAccount(
+            id=bot_account.id,
+            host=bot_account.host,
+        ),
+        ref=uuid4(),
+        smartapp_id=bot_account.id,
+        data={"method": method_name, "params": payload, "type": "smartapp_rpc"},
+        opts={},
+        smartapp_api_version=1,
+        files=[],
+        sender=UserSender(
+            huid=user_info.huid,
+            udid=user_udid,
+            ad_login=user_info.ad_login,
+            ad_domain=user_info.ad_domain,
+            username=user_info.username,
+            is_chat_admin=True,
+            is_chat_creator=True,
+            device=UserDevice(
+                manufacturer=None,
+                device_name=None,
+                os=None,
+                pushes=None,
+                timezone=None,
+                permissions=None,
+                platform=None,
+                platform_package_id=None,
+                app_version=None,
+                locale=None,
+            ),
+        ),
+        chat=Chat(
+            id=chat_id,
+            type=ChatTypes.PERSONAL_CHAT,
+        ),
+        raw_command=None,
+    )
+
+
+class RPCAuth(APIKeyHeader):
+    PATTERN = "([^?=&]+)=([^&]*)"
+
+    async def __call__(self, request: Request) -> RPCAuthConfig:  # type: ignore
+        api_key = request.headers.get(self.model.name)
+        if not api_key:
+            return RPCAuthConfig()
+
+        params = re.findall(self.PATTERN, api_key)  # noqa: WPS110
+        if not params:
+            raise HTTPException(
+                status_code=HTTP_403_FORBIDDEN, detail="Invalid RPC Auth format"
+            )
+        try:
+            config = RPCAuthConfig(**dict(params))
+        except ValidationError as ex:
+            raise HTTPException(status_code=HTTP_403_FORBIDDEN, detail=str(ex))
+
+        return config
+
+
+security = RPCAuth(
+    scheme_name="RPC Auth",
+    name="X-RPC-AUTH",
+    description=DOCS,
+)

app/services/openapi.py

@@ -1,107 +1,83 @@
 """OpenAPI utils."""
-from typing import Any, Dict, List, Optional, Sequence, Union
+from typing import Any, Dict, Sequence, Tuple
 
-from fastapi import routing
 from fastapi.encoders import jsonable_encoder
 from fastapi.openapi.models import OpenAPI
-from fastapi.openapi.utils import get_flat_models_from_routes, get_openapi_path
-from fastapi.utils import get_model_definitions
+from fastapi.openapi.utils import get_openapi
+from fastapi.security.base import SecurityBase
 from pybotx_smartapp_rpc import RPCRouter
 from pybotx_smartapp_rpc.openapi_utils import (
     get_rpc_flat_models_from_routes,
+    get_rpc_model_definitions,
     get_rpc_openapi_path,
 )
 from pydantic.schema import get_model_name_map
 from starlette.routing import BaseRoute
 
+from app.services.execute_rpc import security
+
+
+def get_openapi_security_definitions(
+    security_component: SecurityBase,
+) -> Tuple[Dict[str, Any], Dict[str, Any]]:
+    security_definition = jsonable_encoder(
+        security_component.model,
+        by_alias=True,
+        exclude_none=True,
+    )
+    security_name = security_component.scheme_name
+    security_definitions = {security_name: security_definition}
+    operation_security = {security_name: []}  # type: ignore
+    return security_definitions, operation_security
+
 
 def custom_openapi(
     *,
     title: str,
     version: str,
-    openapi_version: str = "3.0.2",
-    description: Optional[str] = None,
     fastapi_routes: Sequence[BaseRoute],
     rpc_router: RPCRouter,
-    tags: Optional[List[Dict[str, Any]]] = None,
-    servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
-    terms_of_service: Optional[str] = None,
-    contact: Optional[Dict[str, Union[str, Any]]] = None,
-    license_info: Optional[Dict[str, Union[str, Any]]] = None,
+    **kwargs: Any,
 ) -> Dict[str, Any]:
-    info: Dict[str, Any] = {"title": title, "version": version}
-    if description:
-        info["description"] = description
-    if terms_of_service:
-        info["termsOfService"] = terms_of_service
-    if contact:
-        info["contact"] = contact
-    if license_info:
-        info["license"] = license_info
-    output: Dict[str, Any] = {"openapi": openapi_version, "info": info}
-    if servers:
-        output["servers"] = servers
-    components: Dict[str, Dict[str, Any]] = {}
-    paths: Dict[str, Dict[str, Any]] = {}
-    # FastAPI
-    flat_fastapi_models = get_flat_models_from_routes(fastapi_routes)
-    fastapi_model_name_map = get_model_name_map(flat_fastapi_models)
-    fast_api_definitions = get_model_definitions(
-        flat_models=flat_fastapi_models, model_name_map=fastapi_model_name_map
+    openapi_dict = get_openapi(
+        title=title,
+        version=version,
+        routes=fastapi_routes,
+        **kwargs,
     )
 
-    # pybotx RPC
+    paths: Dict[str, Dict[str, Any]] = {}
+
     flat_rpc_models = get_rpc_flat_models_from_routes(rpc_router)
     rpc_model_name_map = get_model_name_map(flat_rpc_models)
-    rpc_definitions = get_model_definitions(
+    rpc_definitions = get_rpc_model_definitions(
         flat_models=flat_rpc_models, model_name_map=rpc_model_name_map
     )
-
-    for route in fastapi_routes:
-        if isinstance(route, routing.APIRoute):
-            result = get_openapi_path(
-                route=route, model_name_map=fastapi_model_name_map
-            )
-            if result:
-                path, security_schemes, path_definitions = result
-                if path:
-                    paths.setdefault(route.path_format, {}).update(path)
-                if security_schemes:
-                    components.setdefault("securitySchemes", {}).update(
-                        security_schemes
-                    )
-                if path_definitions:
-                    fast_api_definitions.update(path_definitions)
+    security_definitions, operation_security = get_openapi_security_definitions(
+        security_component=security
+    )
 
     for method_name in rpc_router.rpc_methods.keys():
         if not rpc_router.rpc_methods[method_name].include_in_schema:
             continue
 
-        result = get_rpc_openapi_path(  # type: ignore
+        path = get_rpc_openapi_path(  # type: ignore
             method_name=method_name,
             route=rpc_router.rpc_methods[method_name],
             model_name_map=rpc_model_name_map,
+            security_scheme=operation_security,
         )
-        if result:
-            path, path_definitions = result  # type: ignore
-            if path:
-                paths.setdefault(method_name, {}).update(path)
+        if path:
+            paths.setdefault(f"/{method_name}", {}).update(path)
 
-            if path_definitions:
-                rpc_definitions.update(path_definitions)
-
-    if fast_api_definitions:
-        components["schemas"] = {
-            k: fast_api_definitions[k] for k in sorted(fast_api_definitions)
-        }
     if rpc_definitions:
-        components.setdefault("schemas", {}).update(
+        openapi_dict.setdefault("components", {}).setdefault("schemas", {}).update(
             {k: rpc_definitions[k] for k in sorted(rpc_definitions)}
         )
-    if components:
-        output["components"] = components
-    output["paths"] = paths
-    if tags:
-        output["tags"] = tags
 
-    return jsonable_encoder(OpenAPI(**output), by_alias=True, exclude_none=True)
+    openapi_dict.setdefault("components", {}).setdefault("securitySchemes", {}).update(
+        security_definitions
+    )
+    openapi_dict.setdefault("paths", {}).update(paths)
+
+    return jsonable_encoder(OpenAPI(**openapi_dict), by_alias=True, exclude_none=True)