diff --git a/docs/source/index.rst b/docs/source/index.rst index a4ab220..19ba4ee 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -36,7 +36,7 @@ Features: - :doc:`intuitive interface ` - :doc:`extensibility ` - :doc:`synchronous and asynchronous client backends ` -- :doc:`popular frameworks integration ` (aiohttp, flask, aio_pika) +- :doc:`popular frameworks integration ` (aiohttp, flask, fastapi, aio_pika) - :doc:`builtin parameter validation ` - :doc:`pytest integration ` - :doc:`openapi schema generation support ` @@ -50,6 +50,7 @@ Extra requirements - `aiohttp `_ - `aio_pika `_ - `flask `_ +- `fastapi `_ - `pydantic `_ - `requests `_ - `httpx `_ diff --git a/docs/source/pjrpc/examples.rst b/docs/source/pjrpc/examples.rst index 90036e9..0f4dea5 100644 --- a/docs/source/pjrpc/examples.rst +++ b/docs/source/pjrpc/examples.rst @@ -88,6 +88,13 @@ httpserver :language: python +fastapi server +---------- + +.. literalinclude:: ../../../examples/fastapi_server.py + :language: python + + middlewares ----------- @@ -163,3 +170,10 @@ aiohttp OpenAPI specification .. literalinclude:: ../../../examples/openapi_aiohttp.py :language: python + + +fastapi OpenAPI specification +----------------------------- + +.. literalinclude:: ../../../examples/openapi_fastapi.py + :language: python diff --git a/docs/source/pjrpc/server.rst b/docs/source/pjrpc/server.rst index d6782b0..056c100 100644 --- a/docs/source/pjrpc/server.rst +++ b/docs/source/pjrpc/server.rst @@ -5,7 +5,8 @@ Server ``pjrpc`` supports popular backend frameworks like `aiohttp `_, -`flask `_ and message brokers like `aio_pika `_. +`flask `_, `FastAPI `_ +and message brokers like `aio_pika `_. Running of aiohttp based JSON-RPC server is a very simple process. Just define methods, add them to the diff --git a/examples/fastapi_server.py b/examples/fastapi_server.py new file mode 100644 index 0000000..b44d7d5 --- /dev/null +++ b/examples/fastapi_server.py @@ -0,0 +1,43 @@ +import logging + +import fastapi + +import pjrpc.server +from pjrpc.server.integration import fastapi as integration + +methods = pjrpc.server.MethodRegistry() + + +@methods.add() +async def sum(a: int, b: int) -> int: + return a + b + + +@methods.add() +async def sub(a: int, b: int) -> int: + return a - b + + +@methods.add() +async def div(a: int, b: int) -> float: + return a / b + + +@methods.add() +async def mult(a: int, b: int) -> int: + return a * b + + +@methods.add() +async def ping() -> None: + logging.info("ping") + + +jsonrpc_app = integration.Application('/api/v1') +jsonrpc_app.add_methods(methods) + +app = fastapi.FastAPI() +app.mount("/rpc", jsonrpc_app.http_app) + +if __name__ == "__main__": + logging.basicConfig(level=logging.INFO) diff --git a/examples/openapi_fastapi.py b/examples/openapi_fastapi.py new file mode 100644 index 0000000..20b30c6 --- /dev/null +++ b/examples/openapi_fastapi.py @@ -0,0 +1,170 @@ +import uuid +from typing import Annotated, Any, Mapping + +import fastapi +import pydantic as pd + +import pjrpc.server.specs.extractors.pydantic +from pjrpc.server.integration import fastapi as integration +from pjrpc.server.specs import extractors, openapi +from pjrpc.server.specs.openapi import ui +from pjrpc.server.validators import pydantic as validators + +methods = pjrpc.server.MethodRegistry( + validator_factory=validators.PydanticValidatorFactory(exclude=integration.is_fastapi_request), + metadata_processors=[ + openapi.MethodSpecificationGenerator( + extractor=extractors.pydantic.PydanticMethodInfoExtractor( + exclude=integration.is_fastapi_request, + ), + ), + ], +) + +credentials = {"admin": "admin"} + + +UserName = Annotated[ + str, + pd.Field(description="User name", examples=["John"]), +] + +UserSurname = Annotated[ + str, + pd.Field(description="User surname", examples=["Doe"]), +] + +UserAge = Annotated[ + int, + pd.Field(description="User age", examples=[25]), +] + +UserId = Annotated[ + uuid.UUID, + pd.Field(description="User identifier", examples=["08b02cf9-8e07-4d06-b569-2c24309c1dc1"]), +] + + +class UserIn(pd.BaseModel, title="User data"): + """ + User registration data. + """ + + name: UserName + surname: UserSurname + age: UserAge + + +class UserOut(UserIn, title="User data"): + """ + Registered user data. + """ + + id: UserId + + +class AlreadyExistsError(pjrpc.server.exceptions.TypedError): + """ + User already registered error. + """ + + CODE = 2001 + MESSAGE = "user already exists" + + +class NotFoundError(pjrpc.server.exceptions.TypedError): + """ + User not found error. + """ + + CODE = 2002 + MESSAGE = "user not found" + + +@methods.add( + pass_context='request', + metadata=[ + openapi.metadata( + summary="Creates a user", + tags=['users'], + errors=[AlreadyExistsError], + ), + ], +) +def add_user(request: fastapi.Request[Mapping[str, Any]], user: UserIn) -> UserOut: + """ + Creates a user. + + :param request: http request + :param object user: user data + :return object: registered user + :raise AlreadyExistsError: user already exists + """ + + +@methods.add( + pass_context='request', + metadata=[ + openapi.metadata( + summary="Returns a user", + tags=['users'], + errors=[NotFoundError], + ), + ], +) +def get_user(request: fastapi.Request[Mapping[str, Any]], user_id: UserId) -> UserOut: + """ + Returns a user. + + :param request: http request + :param object user_id: user id + :return object: registered user + :raise NotFoundError: user not found + """ + + +@methods.add( + pass_context='request', + metadata=[ + openapi.metadata( + summary="Deletes a user", + tags=['users'], + errors=[NotFoundError], + deprecated=True, + ), + ], +) +def delete_user(request: fastapi.Request[Mapping[str, Any]], user_id: UserId) -> None: + """ + Deletes a user. + + :param request: http request + :param object user_id: user id + :raise NotFoundError: user not found + """ + + +class JSONEncoder(pjrpc.server.JSONEncoder): + def default(self, o: Any) -> Any: + if isinstance(o, pd.BaseModel): + return o.model_dump() + if isinstance(o, uuid.UUID): + return str(o) + + return super().default(o) + + +openapi_spec = openapi.OpenAPI( + info=openapi.Info(version="1.0.0", title="User storage"), +) + +jsonrpc_app = integration.Application('/api/v1') +jsonrpc_app.add_methods(methods) + +jsonrpc_app.add_spec(openapi_spec, path='openapi.json') +jsonrpc_app.add_spec_ui('swagger', ui.SwaggerUI(), spec_url='../openapi.json') +jsonrpc_app.add_spec_ui('redoc', ui.ReDoc(), spec_url='../openapi.json') + + +app = fastapi.FastAPI() +app.mount("/rpc", jsonrpc_app.http_app) diff --git a/pjrpc/server/integration/fastapi.py b/pjrpc/server/integration/fastapi.py new file mode 100644 index 0000000..e8ee875 --- /dev/null +++ b/pjrpc/server/integration/fastapi.py @@ -0,0 +1,239 @@ +""" +FastApi JSON-RPC server integration. +""" +import functools as ft +import inspect +import json +from typing import Any, Callable, Iterable, Mapping, Optional, get_origin + +import fastapi +from fastapi.responses import JSONResponse +from fastapi.staticfiles import StaticFiles + +import pjrpc +import pjrpc.server +from pjrpc.server import specs, utils +from pjrpc.server.dispatcher import AsyncExecutor, AsyncMiddlewareType, JSONEncoder, MethodRegistry + + +def is_fastapi_request(idx: int, name: str, annotation: Optional[type[Any]], default: Optional[Any]) -> bool: + if annotation is None: + return False + + annotation_origin = get_origin(annotation) + return inspect.isclass(annotation_origin) and issubclass(annotation_origin, fastapi.Request) + + +FastApiDispatcher = pjrpc.server.AsyncDispatcher[fastapi.Request] + + +class Application: + """ + `FastAPI `_ based JSON-RPC server. + + :param prefix: JSON-RPC handler base path + :param http_app: FastApi application instance + :param status_by_error: a function returns http status code by json-rpc error codes, 200 for all errors by default + """ + + def __init__( + self, + prefix: str = '', + http_app: Optional[fastapi.FastAPI] = None, + status_by_error: Callable[[tuple[int, ...]], int] = lambda codes: 200, + executor: Optional[AsyncExecutor] = None, + json_loader: Callable[..., Any] = json.loads, + json_dumper: Callable[..., str] = json.dumps, + json_encoder: type[JSONEncoder] = JSONEncoder, + json_decoder: Optional[type[json.JSONDecoder]] = None, + middlewares: Iterable[AsyncMiddlewareType[fastapi.Request]] = (), + max_batch_size: Optional[int] = None, + ): + self._prefix = prefix.rstrip('/') + self._http_app = http_app or fastapi.FastAPI() + self._status_by_error = status_by_error + + self._executor: Optional[AsyncExecutor] = executor + self._json_loader: Callable[..., Any] = json_loader + self._json_dumper: Callable[..., str] = json_dumper + self._json_encoder: type[JSONEncoder] = json_encoder + self._json_decoder: Optional[type[json.JSONDecoder]] = json_decoder + self._middlewares: Iterable[AsyncMiddlewareType[fastapi.Request]] = middlewares + self._max_batch_size: Optional[int] = max_batch_size + + self._endpoints: dict[str, FastApiDispatcher] = {} + self._subapps: dict[str, Application] = {} + + @property + def prefix(self) -> str: + return self._prefix + + @property + def http_app(self) -> fastapi.FastAPI: + """ + Fastapi application. + """ + + return self._http_app + + @property + def endpoints(self) -> Mapping[str, FastApiDispatcher]: + """ + JSON-RPC application registered endpoints. + """ + + return self._endpoints + + def add_methods(self, registry: MethodRegistry, endpoint: str = '') -> 'Application': + """ + Adds methods to the provided endpoint. + + :param registry: methods registry + :param endpoint: endpoint path + """ + + dispatcher = self._get_endpoint(endpoint) + dispatcher.add_methods(registry) + return self + + def add_subapp(self, prefix: str, subapp: 'Application') -> None: + """ + Adds sub-application accessible under provided prefix. + + :param prefix: path under which sub-application is accessed. + :param subapp: sub-application instance + """ + + prefix = prefix.rstrip('/') + if not prefix: + raise ValueError("prefix cannot be empty") + + for dispatcher in subapp.endpoints.values(): + dispatcher.add_middlewares(*self._middlewares, before=True) + + self._http_app.mount(utils.join_path(self._prefix, prefix), subapp.http_app) + self._subapps[prefix] = subapp + + def add_spec(self, spec: specs.Specification, endpoint: str = '', path: str = '') -> None: + """ + Adds JSON-RPC specification of the provided endpoint to the provided path. + + :param spec: JSON-RPC specification + :param endpoint: specification endpoint + :param path: path under witch the specification will be accessible. + """ + + self._http_app.router.add_route( + utils.join_path(self._prefix, endpoint, path), + ft.partial(self._get_spec, endpoint=endpoint, spec=spec, path=path), + methods=["GET"], + ) + + def add_spec_ui(self, path: str, ui: specs.BaseUI, spec_url: str) -> None: + """ + Adds JSON-RPC specification ui. + + :param path: path under which ui will be accessible. + :param ui: specification ui instance + :param spec_url: specification url + """ + + ui_app = fastapi.FastAPI() + ui_app.router.add_route( + '/', + ft.partial(self._ui_index_page, ui=ui, spec_url=spec_url), methods=["GET"], + ) + ui_app.router.add_route( + '/index.html', + ft.partial(self._ui_index_page, ui=ui, spec_url=spec_url), methods=["GET"], + ) + ui_app.router.mount('/', StaticFiles(directory=ui.get_static_folder())) + + self._http_app.mount(utils.join_path(self._prefix, path), ui_app) + + def generate_spec(self, spec: specs.Specification, base_path: str = '', endpoint: str = '') -> dict[str, Any]: + """ + Generates JSON-RPC specification of the provided endpoint. + + :param spec: JSON-RPC specification + :param base_path: specification base path + :param endpoint: endpoint the specification is generated for + """ + + app_endpoints = self._endpoints + for prefix, subapp in self._subapps.items(): + for subprefix, dispatcher in subapp.endpoints.items(): + app_endpoints[utils.join_path(prefix, subapp._prefix, subprefix)] = dispatcher + + methods = { + utils.remove_prefix(dispatcher_endpoint, endpoint): dispatcher.registry.values() + for dispatcher_endpoint, dispatcher in app_endpoints.items() + if dispatcher_endpoint.startswith(endpoint) + } + return spec.generate( + root_endpoint=utils.join_path(base_path, endpoint), + methods=methods, + ) + + async def _ui_index_page(self, request: fastapi.Request, ui: specs.BaseUI, spec_url: str) -> fastapi.Response: + return fastapi.Response(content=ui.get_index_page(spec_url), media_type='text/html') + + def _get_endpoint(self, endpoint: str) -> FastApiDispatcher: + endpoint = endpoint.rstrip('/') + + if endpoint not in self._endpoints: + self._endpoints[endpoint] = dispatcher = FastApiDispatcher( + executor=self._executor, + json_loader=self._json_loader, + json_dumper=self._json_dumper, + json_encoder=self._json_encoder, + json_decoder=self._json_decoder, + middlewares=self._middlewares, + max_batch_size=self._max_batch_size, + ) + self._http_app.router.add_route( + utils.join_path(self._prefix, endpoint), + ft.partial(self._rpc_handle, dispatcher=dispatcher), + methods=['POST'], + ) + else: + dispatcher = self._endpoints[endpoint] + + return dispatcher + + async def _get_spec( + self, + request: fastapi.Request, + endpoint: str, + spec: specs.Specification, + path: str, + ) -> fastapi.Response: + base_path = utils.remove_suffix(request.scope['path'], suffix=utils.join_path(endpoint, path)) + schema = self.generate_spec(base_path=base_path, endpoint=endpoint.rstrip('/'), spec=spec) + + return JSONResponse(content=schema) + + async def _rpc_handle(self, http_request: fastapi.Request, dispatcher: FastApiDispatcher) -> fastapi.Response: + """ + Handles JSON-RPC request. + + :param http_request: :py:class:`fastapi.Request` + :returns: :py:class:`fastapi.Response` + """ + + content_type = http_request.headers.get("Content-Type", None) + if content_type not in pjrpc.common.REQUEST_CONTENT_TYPES: + raise fastapi.HTTPException(status_code=fastapi.status.HTTP_415_UNSUPPORTED_MEDIA_TYPE) + + try: + request_body_bytes = await http_request.body() + request_text = request_body_bytes.decode() + except UnicodeDecodeError as e: + raise fastapi.HTTPException(status_code=fastapi.status.HTTP_400_BAD_REQUEST) from e + + response = await dispatcher.dispatch(request_text, context=http_request) + if response is None: + return fastapi.Response() + else: + response_text, error_codes = response + return JSONResponse(status_code=self._status_by_error(error_codes), content=response_text) diff --git a/pyproject.toml b/pyproject.toml index 20a727d..ecedd9d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,7 +22,6 @@ classifiers = [ "Topic :: Software Development :: Libraries", "Topic :: Software Development :: Libraries :: Application Frameworks", "Programming Language :: Python", - "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", @@ -31,7 +30,7 @@ classifiers = [ [tool.poetry.dependencies] -python = ">=3.9,<4.0" +python = ">=3.10,<4.0" aio-pika = { version = ">=8.0", optional = true } aiohttp = { version = ">=3.7", optional = true } flask = { version = ">=2.0.0", optional = true } @@ -40,6 +39,7 @@ openapi-ui-bundles = { version = ">=0.1", optional = true } pydantic = {version = ">=1.10.20", optional = true} requests = { version = ">=2.0", optional = true } werkzeug = { version = ">=2.0", optional = true} +fastapi = { version = ">=0.139.0", optional = true} [tool.poetry.extras] @@ -51,6 +51,7 @@ openapi-ui-bundles = ['openapi-ui-bundles'] pydantic = ['pydantic'] requests = ['requests'] werkzeug = ['werkzeug'] +fastapi = ['fastapi'] [tool.poetry.group.docs]