eXpress SmartApp JSON-RPC library
Project description
BotX-SmartApp-RPC
Библиотека, позволяющая писать смартаппы, используя наш JSONRPC-like протокол
Установка
Используя poetry
:
poetry add pybotx-smartapp-rpc
Добавление RPC методов
- Создайте класс для входящих аргументов:
from pybotx_smartap_rpc import RPCArgsBaseModel
...
class SumArgs(RPCArgsBaseModel):
a: int
b: int
- Создайте RPC метод:
from pybotx_smartapp_rpc import SmartApp, RPCRouter, RPCResultResponse
...
rpc = RPCRouter()
...
@rpc.method("sum")
async def sum(
smartapp: SmartApp, rpc_arguments: SumArgs
) -> RPCResultResponse[int]:
return RPCResultResponse(result=rpc_arguments.a + rpc_arguments.b)
# Так же у метода может не быть аргументов:
@rpc.method("answer")
async def answer(smartapp: SmartApp) -> RPCResultResponse[int]:
return RPCResultResponse(result=42)
- Создайте экземпляр
SmartAppRPC
и подключите роутер из прошлого пункта:
from pybotx_smartapp_rpc import SmartAppRPC
from anywhere import methods
...
smartapp = SmartAppRPC(routers=[methods.rpc])
- Сделайте хендлер для
smartapp_event
и вызывайте в нем хендлер библиотеки
@collector.smartapp_event
async def handle_smartapp_event(event: SmartAppEvent, bot: Bot) -> None:
await smartapp.handle_smartapp_event(event, bot)
Продвинутая работа с библиотекой
- В
RPCResultResponse
можно передаватьbotx.File
файлы.
@rpc.method("get-pdf")
async def get_pdf(
smartapp: SmartApp, rpc_arguments: GetPDFArgs
) -> RPCResultResponse[None]:
...
return RPCResultResponse(result=None, files=[...])
- В
SmartAppRPC
,RPCRouter
иRPCRouter.method
можно передать мидлвари, сначала будут вызваны мидлвари приложения, затем мидлвари роутера и в конце мидлвари метода.
smartapp = SmartAppRPC(..., middlewares=[...])
...
rpc = RPCRouter(middlewares=[...])
...
@rpc.method("sum", middlewares=[...])
RPCArgsBaseModel
это алиас дляpydantic.BaseModel
, вы можете использовать все возможности исходного класса.
from uuid import UUID
...
class DelUserArgs(RPCArgsBaseModel):
# pydantic сериализует входящую строку в UUID
user_huid: UUID
- Через объект
smartapp
, передаваемый в хендлер можно получить доступ кevent
иbot
.
...
@rpc.method("del-user")
async def del_user(
smartapp: SmartApp, rpc_arguments: DelUserArgs
) -> RPCResultResponse[None]:
await smartapp.bot.send_message(
body="Done",
bot_id=smartapp.event.bot.id,
chat_id=smartapp.event.chat.id,
)
...
- Используя метод
smartapp.send_event
можно отправлять RPC ивенты сref: null
.
Это может пригодиться при необходимости отправки уведомления не в ответ на RPC запрос.
@rpc.method("notify-me")
async def notify_me(
smartapp: SmartApp, rpc_arguments: NotifyMeArgs
) -> RPCResultResponse[None]:
...
await smartapp.send_event("notified", files=[notify_file])
...
- Используя метод
smartapp.send_push
можно отправлять пуш уведомлений на клиент. И обновлять счетчик уведомлений на икноке смартапа.
@rpc.method("notify-me")
async def notify_me(
smartapp: SmartApp, rpc_arguments: NotifyMeArgs
) -> RPCResultResponse[None]:
await smartapp.send_push(42, "You have 42 new emails!")
...
- В мидлварях можно создавать новые объекты в
smartapp.state
, чтобы потом использовать их в хендлерах.
async def user_middleware(smartapp: SmartApp, rpc_arguments: RPCArgsBaseModel, call_next: Callable) -> RPCResponse[User]:
smartapp.state.user = await User.get(smartapp.message.user_huid)
return await call_next(smartapp, rpc_arguments)
@rpc.method("get-user-fullname")
async def get_user_fullname(smartapp: SmartApp) -> RPCResultResponse[str]:
return RPCResultResponse(result=smartapp.state.user.fullname)
- Можно выбрасывать пользовательские RPC ошибки, которые будут отправлены как ответ на RPC запрос.
from pybotx_smartapp_rpc import RPCErrorExc, RPCError
...
@rpc.method("return-error")
async def return_error(smartapp: SmartApp, rpc_arguments: RaiseOneErrorArgs) -> None:
# one error
raise RPCErrorExc(
RPCError(
reason="It's error reason",
id="CUSTOM_ERROR",
meta={"args": rpc_arguments.dict()},
)
)
# or list of errors
raise RPCErrorExc(
[
RPCError(
reason="It's error reason",
id="CUSTOM_ERROR",
meta={"args": rpc_arguments.dict()},
),
RPCError(
reason="It's one more error reason",
id="CUSTOM_ERROR_NUMBER_TWO",
meta={"args": rpc_arguments.dict()},
)
]
)
- Можно добавить хендлер на определенный тип исключений. В него будут отправлять исключения того же и дочерних классов.
Хендлер обязан возвращать
RPCErrorResponse
, ошибки из которого будут отправлены источнику запроса.
from pybotx_smartapp_rpc import SmartAppRPC, RPCErrorResponse
...
async def key_error_handler(exc: KeyError, smartapp: SmartApp) -> RPCErrorResponse:
key = exc.args[0]
return RPCErrorResponse(
errors=[
RPCError(
reason=f"Key {key} not found.",
id="KEY_ERROR",
meta={"key": key},
),
]
)
smartapp = SmartAppRPC(..., exception_handlers={KeyError: key_error_handler})
Swagger documentation
Можно подключить rpc роутеры к авто генерируемой документации FastAPI и использовать документацию в Swagger. Для этого необходимо переопределить функцию для генерации OpenAPI схемы:
from fastapi import FastAPI
application = FastAPI()
def get_custom_openapi():
return custom_openapi(
title="Smartapp API",
version="0.1.0",
fastapi_routes=application.routes,
rpc_router=smartapp.router,
)
application.openapi = get_custom_openapi
Пример функции custom_openapi
:
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
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,
) -> 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
)
# pybotx RPC
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(
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)
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(
method_name=method_name,
route=rpc_router.rpc_methods[method_name],
model_name_map=rpc_model_name_map,
)
if result:
path, path_definitions = result
if path:
paths.setdefault(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(
{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) # type: ignore
Возможности RPC Swagger
- Можно добавлять теги к запросам, анaлогично FastAPI.
rpc = RPCRouter(tags=["RPC"])
@rpc.method("documented-method", tags=["docs"])
async def docs(
smartapp: SmartApp, rpc_arguments: DocumentedArgs
) -> RPCResultResponse[DocumentedResponse]:
"""Desctiption of this method."""
...
- Можно переопределять pydantic модель успешного ответа.
@rpc.method("method", return_type=Response)
async def method(
smartapp: SmartApp, rpc_arguments: MethodArgs
) -> RPCResultResponse[int]:
...
- Можно исключать некоторые методы из документации.
rpc = RPCRouter(include_in_schema=False)
@rpc.method("_hidden_method", include_in_schema=False)
async def hidden_method(smartapp: SmartApp) -> RPCResultResponse[int]:
...
- Можно определять пользовательские ошибки.
from pybotx_smartapp_rpc import RPCError, RPCErrorExc
class Meta(BaseModel):
user_id: int
username: str
class UsernotFoundError(RPCError):
id = "UserNotFound"
reason = "User not found in db"
meta: Meta
@rpc.method("method-with_error", errors=[UsernotFoundError])
async def get_user(
smartapp: SmartApp, rpc_arguments: UserArgs
) -> RPCResultResponse[User]:
...
raise RPCErrorExc(UsernotFoundError(meta={"user_id": 1, "username": "test"}))
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
pybotx_smartapp_rpc-0.6.0.tar.gz
(15.9 kB
view hashes)
Built Distribution
Close
Hashes for pybotx_smartapp_rpc-0.6.0.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | b0b42824c55d89e129af90588d39912b2557119997651715f058504ae631343e |
|
MD5 | e3fb92f1f2bae8b8aab5b07f14bd9aeb |
|
BLAKE2b-256 | a012c07286fa4c024d5c525cdae0b2e6fd4aee48f3768987bd88cc27abd4ec92 |
Close
Hashes for pybotx_smartapp_rpc-0.6.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 178665f4b449bb3748132bec086cfde54a9d8ee8048d41257f5489a0402399e4 |
|
MD5 | f58460046798418e1cabdd687d32586d |
|
BLAKE2b-256 | a5117b478f8e0feed49dabb7fa5edabb6ee29170e1b2e574f0868f6d2993b674 |