1
0
bthlabs-jsonrpc/packages/bthlabs-jsonrpc-aiohttp/bthlabs_jsonrpc_aiohttp/views.py
2024-01-15 20:20:10 +00:00

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(),
)