79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
# -*- coding: utf-8 -*-
|
|
# bthlabs-jsonrpc-aiohttp | (c) 2022-present Tomek Wójcik | MIT License
|
|
from __future__ import annotations
|
|
|
|
import inspect
|
|
|
|
from aiohttp import web
|
|
from bthlabs_jsonrpc_core.codecs import Codec, JSONCodec
|
|
|
|
from bthlabs_jsonrpc_aiohttp.executor import AioHttpExecutor
|
|
|
|
|
|
class JSONRPCView:
|
|
"""
|
|
The JSONRPC View. This is the main JSONRPC entry point. Use it to register
|
|
your JSONRPC endpoints.
|
|
|
|
Example:
|
|
|
|
.. code-block:: python
|
|
|
|
from bthlabs_jsonrpc_aiohttp import JSONRPCView
|
|
|
|
app.add_routes([
|
|
web.post('/rpc', JSONRPCView()),
|
|
web.post('/example/rpc', JSONRPCView(namespace='example')),
|
|
])
|
|
"""
|
|
|
|
def __init__(self,
|
|
namespace: str | None = None,
|
|
codec: type[Codec] | None = None):
|
|
self.namespace: str | None = namespace
|
|
self.codec: type[Codec] = codec or JSONCodec
|
|
|
|
# pragma mark - Public interface
|
|
|
|
async def can_call(self,
|
|
request: web.Request,
|
|
method: str,
|
|
args: list,
|
|
kwargs: dict) -> bool:
|
|
"""
|
|
Hook for subclasses to perform additional per-call permissions checks
|
|
etc. The default implementation returns ``True``.
|
|
"""
|
|
return True
|
|
|
|
async def get_codec(self, request: web.Request) -> Codec:
|
|
"""Returns a codec configured for the *request*."""
|
|
return self.codec()
|
|
|
|
async def get_executor(self, request: web.Request) -> AioHttpExecutor:
|
|
"""Returns an executor configured for the *request*."""
|
|
codec = await self.get_codec(request)
|
|
|
|
return AioHttpExecutor(
|
|
request, self.can_call, namespace=self.namespace, codec=codec,
|
|
)
|
|
|
|
async def __call__(self, request: web.Request) -> web.Response:
|
|
"""The request handler."""
|
|
executor = await self.get_executor(request)
|
|
|
|
serializer = await executor.execute()
|
|
if serializer is None:
|
|
return web.Response(body='')
|
|
|
|
codec = await self.get_codec(request)
|
|
|
|
body = codec.encode(serializer.data)
|
|
if inspect.isawaitable(body):
|
|
body = await body
|
|
|
|
return web.Response(
|
|
body=body,
|
|
content_type=codec.get_content_type(),
|
|
)
|