Server#

Misc#

JSON-RPC server package.

class pjrpc.server.AsyncDispatcher(*, request_class=<class 'pjrpc.common.v20.Request'>, response_class=<class 'pjrpc.common.v20.Response'>, batch_request=<class 'pjrpc.common.v20.BatchRequest'>, batch_response=<class 'pjrpc.common.v20.BatchResponse'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.server.dispatcher.JSONEncoder'>, json_decoder=None, middlewares=(), error_handlers={}, max_batch_size=None, concurrent_batch=True)[source]#

Bases: BaseDispatcher

Asynchronous method dispatcher.

Parameters:
async dispatch(request_text, context=None)[source]#

Deserializes request, dispatches it to the required method and serializes the result.

Parameters:

  • request_text (str) – request text representation

  • context (Optional[Any]) – application context (if supported)

Returns:

response text representation

Return type:

Optional[Tuple[str, Tuple[int, …]]]

class pjrpc.server.BaseDispatcher(*, request_class=<class 'pjrpc.common.v20.Request'>, response_class=<class 'pjrpc.common.v20.Response'>, batch_request=<class 'pjrpc.common.v20.BatchRequest'>, batch_response=<class 'pjrpc.common.v20.BatchResponse'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.server.dispatcher.JSONEncoder'>, json_decoder=None, middlewares=(), error_handlers={})[source]#

Bases: object

Method dispatcher.

Parameters:
add(method, name=None, context=None)[source]#

Adds method to the registry.

Parameters:
Return type:

None

add_methods(*methods)[source]#

Adds methods to the registry.

Parameters:

methods (Union[MethodRegistry, Method, Callable[[...], Any]]) – method list. Each method may be an instance of pjrpc.server.MethodRegistry, pjrpc.server.Method or plain function

Return type:

None

view(view)[source]#

Adds class based view to the registry.

Parameters:

view (Type[ViewMixin]) – view to be added

Return type:

None

class pjrpc.server.Dispatcher(*, request_class=<class 'pjrpc.common.v20.Request'>, response_class=<class 'pjrpc.common.v20.Response'>, batch_request=<class 'pjrpc.common.v20.BatchRequest'>, batch_response=<class 'pjrpc.common.v20.BatchResponse'>, json_loader=<function loads>, json_dumper=<function dumps>, json_encoder=<class 'pjrpc.server.dispatcher.JSONEncoder'>, json_decoder=None, middlewares=(), error_handlers={}, max_batch_size=None)[source]#

Bases: BaseDispatcher

Synchronous method dispatcher.

Parameters:
dispatch(request_text, context=None)[source]#

Deserializes request, dispatches it to the required method and serializes the result.

Parameters:

  • request_text (str) – request text representation

  • context (Optional[Any]) – application context (if supported)

Returns:

response text representation

Return type:

Optional[Tuple[str, Tuple[int, …]]]

class pjrpc.server.JSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: JSONEncoder

Server JSON encoder. All custom server encoders should be inherited from it.

default(o)[source]#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
Parameters:

o (Any) –

Return type:

Any

class pjrpc.server.Method(method, name=None, context=None, positional=False)[source]#

Bases: object

JSON-RPC method wrapper. Stores method itself and some metainformation.

Parameters:
class pjrpc.server.MethodRegistry(prefix=None)[source]#

Bases: object

Method registry.

Parameters:

prefix (Optional[str]) – method name prefix to be used for naming containing methods

get(item)[source]#

Returns a method from the registry by name.

Parameters:

item (str) – method name

Returns:

found method or None

Return type:

Optional[Method]

add(maybe_method=None, name=None, context=None, positional=False)[source]#

Decorator adding decorated method to the registry.

Parameters:

  • maybe_method (Optional[Callable[[...], Any]]) – method or None

  • name (Optional[str]) – method name to be used instead of __name__ attribute

  • context (Optional[Any]) – parameter name to be used as an application context

  • positional (bool) – pass context as first positional argument

Returns:

decorated method or decorator

Return type:

Callable[[…], Any]

add_methods(*methods)[source]#

Adds methods to the registry.

Parameters:

methods (Union[Callable[[...], Any], Method]) – methods to be added. Each one can be an instance of pjrpc.server.Method or plain method

Return type:

None

view(maybe_view=None, context=None, prefix=None)[source]#

Methods view decorator.

Parameters:
Returns:

decorator or decorated view

Return type:

Union[ViewMixin, Callable[[…], Any]]

merge(other)[source]#

Merges two registries.

Parameters:

other (MethodRegistry) – registry to be merged in the current one

Return type:

None

class pjrpc.server.ViewMixin(context=None)[source]#

Bases: object

Simple class based method handler mixin. Exposes all public methods.

Parameters:

context (Optional[Any]) –

Types#

pjrpc.server.typedefs.AsyncErrorHandlerType#

Asynchronous server error handler

alias of Callable[[Request, Optional[Any], JsonRpcError], Awaitable[JsonRpcError]]

pjrpc.server.typedefs.AsyncMiddlewareType#

Asynchronous middleware type

alias of Callable[[Request, Optional[Any], Callable[[Request, Optional[Any]], Awaitable[Union[UnsetType, Response]]]], Awaitable[Union[UnsetType, Response]]]

pjrpc.server.typedefs.AsyncHandlerType#

Async RPC handler method, passed to middlewares

alias of Callable[[Request, Optional[Any]], Awaitable[Union[UnsetType, Response]]]

pjrpc.server.typedefs.MiddlewareResponse#

middlewares and handlers return Response or UnsetType

alias of Union[UnsetType, Response]

pjrpc.server.typedefs.MiddlewareType#

Synchronous middleware type

alias of Callable[[Request, Optional[Any], Callable[[Request, Optional[Any]], Union[UnsetType, Response]]], Union[UnsetType, Response]]

pjrpc.server.typedefs.ErrorHandlerType#

Synchronous server error handler

alias of Callable[[Request, Optional[Any], JsonRpcError], JsonRpcError]

pjrpc.server.typedefs.ExcludeFunc#

Parameter exclude function

alias of Callable[[str, Optional[Type[Any]], Optional[Any]], bool]

pjrpc.server.typedefs.ResponseOrUnset#

Return value of RPC handlers and middlewares

alias of Union[UnsetType, Response]

pjrpc.server.typedefs.ContextType#

Context argument for RPC methods and middlewares

alias of Optional[Any]

Integrations#

aiohttp#

aiohttp JSON-RPC server integration.

class pjrpc.server.integration.aiohttp.Application(path='', spec=None, specs=(), app=None, status_by_error=<function Application.<lambda>>, **kwargs)[source]#

Bases: object

aiohttp based JSON-RPC server.

Parameters:
property app: Application#

aiohttp application.

property dispatcher: AsyncDispatcher#

JSON-RPC method dispatcher.

property endpoints: Mapping[str, AsyncDispatcher]#

JSON-RPC application registered endpoints.

add_subapp(prefix, subapp)[source]#

Adds sub-application.

Parameters:

  • prefix (str) – sub-application prefix

  • subapp (Application) – sub-application to be added

Return type:

None

add_endpoint(prefix, subapp=None, **kwargs)[source]#

Adds additional endpoint.

Parameters:

  • prefix (str) – endpoint prefix

  • subapp (Optional[Application]) – aiohttp subapp the endpoint will be served on

  • kwargs (Any) – arguments to be passed to the dispatcher pjrpc.server.Dispatcher

Returns:

dispatcher

Return type:

AsyncDispatcher

flask#

Flask JSON-RPC extension.

class pjrpc.server.integration.flask.JsonRPC(path, spec=None, specs=(), status_by_error=<function JsonRPC.<lambda>>, **kwargs)[source]#

Bases: object

Flask framework JSON-RPC extension class.

Parameters:
property dispatcher: Dispatcher#

JSON-RPC method dispatcher.

property endpoints: Dict[str, Dispatcher]#

JSON-RPC application registered endpoints.

add_endpoint(prefix, blueprint=None, **kwargs)[source]#

Adds additional endpoint.

Parameters:

  • prefix (str) – endpoint prefix

  • blueprint (Optional[Blueprint]) – flask blueprint the endpoint will be served on

  • kwargs (Any) – arguments to be passed to the dispatcher pjrpc.server.Dispatcher

Returns:

dispatcher

Return type:

Dispatcher

init_app(app)[source]#

Initializes flask application with JSON-RPC extension.

Parameters:

app (Union[Flask, Blueprint]) – flask application instance

Return type:

None

kombu#

kombu JSON-RPC server integration.

class pjrpc.server.integration.kombu.Executor(broker_url, queue_name, conn_args=None, queue_args=None, publish_args=None, prefetch_count=0, **kwargs)[source]#

Bases: ConsumerProducerMixin

kombu based JSON-RPC server.

Parameters:

  • broker_url (str) – broker connection url

  • queue_name (str) – requests queue name

  • conn_args (Optional[Dict[str, Any]]) – additional connection arguments

  • queue_args (Optional[Dict[str, Any]]) – queue arguments

  • publish_args (Optional[Dict[str, Any]]) – message publish additional arguments

  • prefetch_count (int) – worker prefetch count

  • kwargs (Any) – dispatcher additional arguments

property dispatcher: Dispatcher#

JSON-RPC method dispatcher.

aio_pika#

class pjrpc.server.integration.aio_pika.Executor(broker_url, rx_queue_name, tx_exchange_name=None, tx_routing_key=None, prefetch_count=0, **kwargs)[source]#

Bases: object

aio_pika based JSON-RPC server.

Parameters:

  • broker_url (URL) – broker connection url

  • rx_queue_name (str) – requests queue name

  • tx_exchange_name (Optional[str]) – response exchange name

  • tx_routing_key (Optional[str]) – response routing key

  • prefetch_count (int) – worker prefetch count

  • kwargs (Any) – dispatcher additional arguments

property dispatcher: AsyncDispatcher#

JSON-RPC method dispatcher.

async start(queue_args=None, exchange_args=None)[source]#

Starts executor.

Parameters:
Return type:

None

async shutdown()[source]#

Stops executor.

Return type:

None

werkzeug#

class pjrpc.server.integration.werkzeug.JsonRPC(path='', **kwargs)[source]#

Bases: object

werkzeug server JSON-RPC integration.

Parameters:
property dispatcher: Dispatcher#

JSON-RPC method dispatcher.

Validators#

JSON-RPC method parameters validators.

class pjrpc.server.validators.BaseValidator(exclude_param=None)[source]#

Bases: object

Base method parameters validator. Uses inspect.signature() for validation.

Parameters:

exclude_param (Optional[Callable[[str, Optional[Type[Any]], Optional[Any]], bool]]) – a function that decides if the parameters must be excluded from validation (useful for dependency injection)

validate(maybe_method=None, **kwargs)[source]#

Decorator marks a method the parameters of which to be validated when calling it using JSON-RPC protocol.

Parameters:

  • maybe_method (Optional[Callable[[...], Any]]) – method the parameters of which to be validated or None if called as @validate(…)

  • kwargs (Any) – validator arguments

Return type:

Callable[[…], Any]

validate_method(method, params, exclude=(), **kwargs)[source]#

Validates params against method signature.

Parameters:
Raises:

pjrpc.server.validators.ValidationError

Returns:

bound method parameters

Return type:

Dict[str, Any]

bind(signature, params)[source]#

Binds parameters to method. :param signature: method to bind parameters to :param params: parameters to be bound

Raises:

ValidationError is parameters binding failed

Returns:

bound parameters

Parameters:
Return type:

BoundArguments

signature(method, exclude)[source]#

Returns method signature.

Parameters:

  • method (Callable[[...], Any]) – method to get signature of

  • exclude (Tuple[str, ...]) – parameters to be excluded

Returns:

signature

Return type:

Signature

exception pjrpc.server.validators.ValidationError[source]#

Bases: Exception

Method parameters validation error. Raised when parameters validation failed.

jsonschema#

class pjrpc.server.validators.jsonschema.JsonSchemaValidator(exclude_param=None, **kwargs)[source]#

Bases: BaseValidator

Parameters validator based on jsonschema library.

Parameters:
validate_method(method, params, exclude=(), **kwargs)[source]#

Validates params against method using pydantic validator.

Parameters:
Raises:

pjrpc.server.validators.ValidationError

Return type:

Dict[str, Any]

pydantic#

class pjrpc.server.validators.pydantic.PydanticValidator(coerce=True, exclude_param=None, **config_args)[source]#

Bases: BaseValidator

Parameters validator based on pydantic library. Uses python type annotations for parameters validation.

Parameters:

  • coerce (bool) – if True returns converted (coerced) parameters according to parameter type annotation otherwise returns parameters as is

  • exclude_param (Optional[Callable[[str, Optional[Type[Any]], Optional[Any]], bool]]) – a function that decides if the parameters must be excluded from validation (useful for dependency injection)

  • config_args (Any) –

validate_method(method, params, exclude=(), **kwargs)[source]#

Validates params against method using pydantic validator.

Parameters:
Returns:

coerced parameters if coerce flag is True otherwise parameters as is

Raises:

ValidationError

Return type:

Dict[str, Any]

build_validation_schema(signature)[source]#

Builds pydantic model based validation schema from method signature.

Parameters:

signature (Signature) – method signature to build schema for

Returns:

validation schema

Return type:

Dict[str, Any]

Specification#

class pjrpc.server.specs.JSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]#

Bases: JSONEncoder

Schema JSON encoder.

default(o)[source]#

Implement this method in a subclass such that it returns a serializable object for o, or calls the base implementation (to raise a TypeError).

For example, to support arbitrary iterators, you could implement default like this:

def default(self, o):
    try:
        iterable = iter(o)
    except TypeError:
        pass
    else:
        return list(iterable)
    # Let the base class default method raise the TypeError
    return JSONEncoder.default(self, o)
Parameters:

o (Any) –

Return type:

Any

class pjrpc.server.specs.BaseUI[source]#

Bases: ABC

Base UI.

abstract get_static_folder()[source]#

Returns ui statics folder.

Return type:

str

abstract get_index_page(spec_url)[source]#

Returns ui index webpage.

Parameters:

spec_url (str) – specification url.

Return type:

str

class pjrpc.server.specs.Specification(path='/spec.json', ui=None, ui_path=None)[source]#

Bases: ABC

JSON-RPC specification.

Parameters:

  • path (str) – specification url path suffix

  • ui (Optional[BaseUI]) – specification ui instance

  • ui_path (Optional[str]) – specification ui url path suffix

property path: str#

Returns specification url path.

property ui: Optional[BaseUI]#

Returns ui instance.

property ui_path: Optional[str]#

Returns specification ui url path.

abstract schema(path, methods_map={})[source]#

Returns specification schema.

Parameters:

  • path (str) – methods endpoint path

  • methods_map (Mapping[str, Iterable[Method]]) – methods map the specification is generated for. Each item is a mapping from a prefix to methods on which the methods will be served

Return type:

Dict[str, Any]

extractors#

class pjrpc.server.specs.extractors.BaseSchemaExtractor[source]#

Bases: object

Base method schema extractor.

extract_params_schema(method_name, method, ref_template, exclude=())[source]#

Extracts params schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_request_schema(method_name, method, ref_template, exclude=())[source]#

Extracts request schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_result_schema(method_name, method, ref_template)[source]#

Extracts result schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_response_schema(method_name, method, ref_template, errors=None)[source]#

Extracts response schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_error_response_schema(method_name, method, ref_template, errors=None)[source]#

Extracts error response schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_description(method)[source]#

Extracts method description.

Parameters:

method (Callable[[...], Any]) –

Return type:

Union[UnsetType, str]

extract_summary(method)[source]#

Extracts method summary.

Parameters:

method (Callable[[...], Any]) –

Return type:

Union[UnsetType, str]

extract_errors(method)[source]#

Extracts method errors.

Parameters:

method (Callable[[...], Any]) –

Return type:

Union[UnsetType, List[Type[JsonRpcError]]]

extract_deprecation_status(method)[source]#

Extracts method deprecation status.

Parameters:

method (Callable[[...], Any]) –

Return type:

Union[UnsetType, bool]

class pjrpc.server.specs.extractors.pydantic.JsonRpcRequest(*, jsonrpc, id=None, method, params)[source]#

Bases: BaseModel, Generic[MethodT, ParamsT]

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'strict': True, 'title': 'Request'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.JsonRpcRequestWrapper(root=PydanticUndefined)[source]#

Bases: RootModel[~RequestT], Generic[RequestT]

Parameters:

root (RootModelRootType) –

model_config: ClassVar[ConfigDict] = {'title': 'Request'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.JsonRpcResponseSuccess(*, jsonrpc, id, result)[source]#

Bases: BaseModel, Generic[ResultT]

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'strict': True, 'title': 'Success'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.JsonRpcError(*, code, message, data)[source]#

Bases: BaseModel, Generic[ErrorCodeT, ErrorDataT]

Parameters:

  • code (ErrorCodeT) –

  • message (str) –

  • data (ErrorDataT) –

model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'strict': True, 'title': 'Error'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.JsonRpcResponseError(*, jsonrpc, id, error)[source]#

Bases: BaseModel, Generic[JsonRpcErrorT]

Parameters:
model_config: ClassVar[ConfigDict] = {'extra': 'forbid', 'strict': True, 'title': 'ResponseError'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.JsonRpcResponseWrapper(root=PydanticUndefined)[source]#

Bases: RootModel[~ResponseT], Generic[ResponseT]

Parameters:

root (RootModelRootType) –

model_config: ClassVar[ConfigDict] = {'title': 'Response'}#

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pjrpc.server.specs.extractors.pydantic.PydanticSchemaExtractor(exclude_param=None, **config_args)[source]#

Bases: BaseSchemaExtractor

Pydantic method specification extractor.

Parameters:
extract_params_schema(method_name, method, ref_template, exclude=())[source]#

Extracts params schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_request_schema(method_name, method, ref_template, exclude=())[source]#

Extracts request schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_result_schema(method_name, method, ref_template)[source]#

Extracts result schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_response_schema(method_name, method, ref_template, errors=None)[source]#

Extracts response schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

extract_error_response_schema(method_name, method, ref_template, errors=None)[source]#

Extracts error response schema.

Parameters:
Return type:

Tuple[Dict[str, Any], Dict[str, Dict[str, Any]]]

schemas#

OpenAPI Specification generator. See https://swagger.io/specification/.

class pjrpc.server.specs.openapi.Reference(ref, summary=UNSET, description=UNSET)[source]#

Bases: object

A simple object to allow referencing other components in the OpenAPI document, internally and externally.

Parameters:

  • ref (str) – the reference identifier.

  • summary (Union[UnsetType, str]) – a short summary which by default SHOULD override that of the referenced component.

  • description (Union[UnsetType, str]) – a description which by default SHOULD override that of the referenced component

class pjrpc.server.specs.openapi.Contact(name=UNSET, url=UNSET, email=UNSET)[source]#

Bases: object

Contact information for the exposed API.

Parameters:
class pjrpc.server.specs.openapi.License(name, identifier=UNSET, url=UNSET)[source]#

Bases: object

License information for the exposed API.

Parameters:

  • name (str) – the license name used for the API

  • identifier (Union[UnsetType, str]) – an SPDX license expression for the API

  • url (Union[UnsetType, str]) – a URL to the license used for the API

class pjrpc.server.specs.openapi.Info(title, version, summary=UNSET, description=UNSET, contact=UNSET, license=UNSET, termsOfService=UNSET)[source]#

Bases: object

Metadata about the API.

Parameters:
class pjrpc.server.specs.openapi.ServerVariable(default, enum=UNSET, description=UNSET)[source]#

Bases: object

An object representing a Server Variable for server URL template substitution.

Parameters:

  • default (str) – the default value to use for substitution, which SHALL be sent if an alternate value is not supplied

  • enum (Union[UnsetType, List[str]]) – an enumeration of string values to be used if the substitution options are from a limited set

  • description (Union[UnsetType, str]) – an optional description for the server variable

class pjrpc.server.specs.openapi.Server(url, description=UNSET, variables=UNSET)[source]#

Bases: object

Connectivity information of a target server.

Parameters:

  • url (str) – a URL to the target host

  • description (Union[UnsetType, str]) – an optional string describing the host designated by the URL

  • variables (Union[UnsetType, Dict[str, ServerVariable]]) – a map between a variable name and its value. The value is used for substitution in the server’s URL template.

Bases: object

The Link object represents a possible design-time link for a response.

Parameters:

  • operationRef (Union[UnsetType, str]) – a relative or absolute URI reference to an OAS operation

  • operationId (Union[UnsetType, str]) – the name of an existing, resolvable OAS operation, as defined with a unique operationId

  • parameters (Union[UnsetType, Dict[str, Any]]) – a map representing parameters to pass to an operation as specified with operationId or identified via operationRef

  • requestBody (Union[UnsetType, Any]) – a literal value or {expression} to use as a request body when calling the target operation

  • description (Union[UnsetType, str]) – a description of the link

  • server (Union[UnsetType, Server]) – a server object to be used by the target operation.

class pjrpc.server.specs.openapi.ExternalDocumentation(url, description=UNSET)[source]#

Bases: object

Allows referencing an external resource for extended documentation.

Parameters:

  • url (str) – a short description of the target documentation.

  • description (Union[UnsetType, str]) – the URL for the target documentation

class pjrpc.server.specs.openapi.Tag(name, description=UNSET, externalDocs=UNSET)[source]#

Bases: object

A list of tags for API documentation control. Tags can be used for logical grouping of methods by resources or any other qualifier.

Parameters:
class pjrpc.server.specs.openapi.SecuritySchemeType(value)[source]#

Bases: str, Enum

The type of the security scheme.

class pjrpc.server.specs.openapi.ApiKeyLocation(value)[source]#

Bases: str, Enum

The location of the API key.

class pjrpc.server.specs.openapi.OAuthFlow(authorizationUrl, tokenUrl, scopes, refreshUrl=UNSET)[source]#

Bases: object

Configuration details for a supported OAuth Flow.

Parameters:

  • authorizationUrl (str) – the authorization URL to be used for this flow

  • tokenUrl (str) – the token URL to be used for this flow

  • refreshUrl (Union[UnsetType, str]) – the URL to be used for obtaining refresh tokens

  • scopes (Dict[str, str]) – the available scopes for the OAuth2 security scheme

class pjrpc.server.specs.openapi.OAuthFlows(implicit=UNSET, password=UNSET, clientCredentials=UNSET, authorizationCode=UNSET)[source]#

Bases: object

Configuration of the supported OAuth Flows.

Parameters:
class pjrpc.server.specs.openapi.SecurityScheme(type, scheme=UNSET, name=UNSET, location=UNSET, bearerFormat=UNSET, flows=UNSET, openIdConnectUrl=UNSET, description=UNSET)[source]#

Bases: object

Defines a security scheme that can be used by the operations.

Parameters:
class pjrpc.server.specs.openapi.MethodExample(params, result, version='2.0', summary=UNSET, description=UNSET)[source]#

Bases: object

Method usage example.

Parameters:

  • params (Dict[str, Any]) – example parameters

  • result (Any) – example result

  • name – name for the example pairing

  • summary (Union[UnsetType, str]) – short description for the example pairing

  • description (Union[UnsetType, str]) – a verbose explanation of the example pairing

  • version (str) –

class pjrpc.server.specs.openapi.ExampleObject(value=UNSET, summary=UNSET, description=UNSET, externalValue=UNSET)[source]#

Bases: object

Method usage example.

Parameters:
class pjrpc.server.specs.openapi.Encoding(contentType=UNSET, headers=UNSET, style=UNSET, explode=UNSET, allowReserved=UNSET)[source]#

Bases: object

A single encoding definition applied to a single schema property.

Parameters:

  • contentType (Union[UnsetType, str]) – the Content-Type for encoding a specific property

  • headers (Union[UnsetType, Dict[str, Union[Parameter, Reference]]]) – a map allowing additional information to be provided as headers

  • style (Union[UnsetType, str]) – describes how a specific property value will be serialized depending on its type

  • explode (Union[UnsetType, bool]) – when this is true, property values of type array or object generate separate parameters for each value of the array, or key-value-pair of the map

  • allowReserved (Union[UnsetType, bool]) – determines whether the parameter value SHOULD allow reserved characters, as defined by RFC3986 to be included without percent-encoding

class pjrpc.server.specs.openapi.ParameterLocation(value)[source]#

Bases: str, Enum

The location of the parameter.

class pjrpc.server.specs.openapi.StyleType(value)[source]#

Bases: str, Enum

Describes how the parameter value will be serialized depending on the type of the parameter value.

class pjrpc.server.specs.openapi.MediaType(schema=UNSET, example=UNSET, examples=UNSET, encoding=UNSET)[source]#

Bases: object

Each Media Type Object provides schema and examples for the media type identified by its key.

Parameters:
class pjrpc.server.specs.openapi.RequestBody(content, required=UNSET, description=UNSET)[source]#

Bases: object

Describes a single request body.

Parameters:
class pjrpc.server.specs.openapi.Parameter(name, location, description=UNSET, required=UNSET, deprecated=UNSET, allowEmptyValue=UNSET, style=UNSET, explode=UNSET, allowReserved=UNSET, schema=UNSET, example=UNSET, examples=UNSET, content=UNSET)[source]#

Bases: object

Describes a single operation parameter.

Parameters:

  • name (str) – the name of the parameter

  • location (ParameterLocation) – the location of the parameter

  • description (Union[UnsetType, str]) – a brief description of the parameter

  • required (Union[UnsetType, bool]) – determines whether this parameter is mandatory

  • deprecated (Union[UnsetType, bool]) – a parameter is deprecated and SHOULD be transitioned out of usage

  • allowEmptyValue (Union[UnsetType, bool]) – the ability to pass empty-valued parameters

  • style (Union[UnsetType, StyleType]) – describes how the parameter value will be serialized depending on the type of the parameter value

  • explode (Union[UnsetType, bool]) – when this is true, parameter values of type array or object generate separate parameters for each value of the array or key-value pair of the map

  • allowReserved (Union[UnsetType, bool]) – determines whether the parameter value SHOULD allow reserved characters, as defined by RFC3986 :/?#[]@!$&’()*+,;= to be included without percent-encoding

  • schema (Union[UnsetType, Dict[str, Any]]) – the schema defining the type used for the parameter.

  • examples (Union[UnsetType, Dict[str, ExampleObject]]) – examples of the parameter’s potential value

  • content (Union[UnsetType, Dict[str, MediaType]]) – a map containing the representations for the parameter

  • example (Union[UnsetType, Any]) –

pjrpc.server.specs.openapi.Header#

alias of Parameter

class pjrpc.server.specs.openapi.Response(description, headers=UNSET, content=UNSET, links=UNSET)[source]#

Bases: object

A container for the expected responses of an operation.

Parameters:
class pjrpc.server.specs.openapi.Operation(responses=UNSET, requestBody=UNSET, tags=UNSET, summary=UNSET, description=UNSET, externalDocs=UNSET, operationId=UNSET, deprecated=UNSET, servers=UNSET, security=UNSET, parameters=UNSET)[source]#

Bases: object

Describes a single API operation on a path.

Parameters:
class pjrpc.server.specs.openapi.Path(get=UNSET, put=UNSET, post=UNSET, delete=UNSET, options=UNSET, head=UNSET, patch=UNSET, trace=UNSET, summary=UNSET, description=UNSET, servers=UNSET, parameters=UNSET)[source]#

Bases: object

Describes the interface for the given method name.

Parameters:
class pjrpc.server.specs.openapi.Components(schemas=UNSET, responses=UNSET, parameters=UNSET, examples=UNSET, requestBodies=UNSET, headers=UNSET, securitySchemes=UNSET, links=UNSET, pathItems=UNSET)[source]#

Bases: object

Holds a set of reusable objects for different aspects of the OAS.

Parameters:
class pjrpc.server.specs.openapi.SpecRoot(info, paths, components, servers=UNSET, externalDocs=UNSET, tags=UNSET, security=UNSET, openapi='3.1.0', jsonSchemaDialect=UNSET)[source]#

Bases: object

The root object of the OpenAPI description.

Parameters:

  • info (Info) – provides metadata about the API

  • servers (Union[UnsetType, List[Server]]) – an array of Server Objects, which provide connectivity information to a target server

  • externalDocs (Union[UnsetType, ExternalDocumentation]) – additional external documentation

  • openapi (str) – the semantic version number of the OpenAPI Specification version that the OpenAPI document uses

  • jsonSchemaDialect (Union[UnsetType, str]) – the default value for the $schema keyword within Schema Objects

  • tags (Union[UnsetType, List[Tag]]) – a list of tags used by the specification with additional metadata

  • security (Union[UnsetType, List[Dict[str, List[str]]]]) – a declaration of which security mechanisms can be used across the API

  • paths (Dict[str, Path]) –

  • components (Components) –

class pjrpc.server.specs.openapi.OpenApiMeta[source]#

Bases: TypedDict

pjrpc.server.specs.openapi.annotate(params_schema=UNSET, result_schema=UNSET, errors=UNSET, examples=UNSET, tags=UNSET, summary=UNSET, description=UNSET, external_docs=UNSET, deprecated=UNSET, security=UNSET, parameters=UNSET, servers=UNSET, component_name_prefix=None)[source]#

Adds Open Api specification annotation to the method.

Parameters:
Return type:

Callable[[Func], Func]

class pjrpc.server.specs.openapi.OpenAPI(info, path='/openapi.json', servers=UNSET, external_docs=UNSET, tags=UNSET, security=UNSET, security_schemes=UNSET, openapi='3.1.0', json_schema_dialect=UNSET, schema_extractor=None, schema_extractors=(), ui=None, ui_path='/ui/', error_http_status_map={})[source]#

Bases: Specification

OpenAPI Specification.

Parameters:
schema(path, methods_map={}, component_name_prefix='')[source]#

Returns specification schema.

Parameters:

  • path (str) – methods endpoint path

  • methods_map (Mapping[str, Iterable[Method]]) – methods map the specification is generated for. Each item is a mapping from a prefix to methods on which the methods will be served

  • component_name_prefix (str) –

Return type:

Dict[str, Any]

class pjrpc.server.specs.openapi.SwaggerUI(**configs)[source]#

Bases: BaseUI

Swagger UI.

Parameters:
get_static_folder()[source]#

Returns ui statics folder.

Return type:

str

get_index_page(spec_url)[source]#

Returns ui index webpage.

Parameters:

spec_url (str) – specification url.

Return type:

str

class pjrpc.server.specs.openapi.RapiDoc(**configs)[source]#

Bases: BaseUI

RapiDoc UI.

Parameters:

  • config – documentation configurations (see https://mrin9.github.io/RapiDoc/api.html). Be aware that configuration parameters should be in snake case, for example: parameter heading-text should be passed as heading_text)

  • configs (Any) –

get_static_folder()[source]#

Returns ui statics folder.

Return type:

str

get_index_page(spec_url)[source]#

Returns ui index webpage.

Parameters:

spec_url (str) – specification url.

Return type:

str

class pjrpc.server.specs.openapi.ReDoc(**configs)[source]#

Bases: BaseUI

ReDoc UI.

Parameters:

  • config – documentation configurations (see https://github.com/Redocly/redoc#configuration). Be aware that configuration parameters should be in snake case, for example: parameter heading-text should be passed as heading_text)

  • configs (Any) –

get_static_folder()[source]#

Returns ui statics folder.

Return type:

str

get_index_page(spec_url)[source]#

Returns ui index webpage.

Parameters:

spec_url (str) – specification url.

Return type:

str

OpenRPC specification generator. See https://spec.open-rpc.org/.

class pjrpc.server.specs.openrpc.Reference(ref)[source]#

Bases: object

A simple object to allow referencing other components in the specification, internally and externally. :param ref: the reference identifier.

Parameters:

ref (str) –

class pjrpc.server.specs.openrpc.Contact(name=UNSET, url=UNSET, email=UNSET)[source]#

Bases: object

Contact information for the exposed API.

Parameters:
class pjrpc.server.specs.openrpc.License(name, url=UNSET)[source]#

Bases: object

License information for the exposed API.

Parameters:

  • name (str) – the license name used for the API

  • url (Union[UnsetType, str]) – a URL to the license used for the API

class pjrpc.server.specs.openrpc.Info(title, version, description=UNSET, contact=UNSET, license=UNSET, termsOfService=UNSET)[source]#

Bases: object

Metadata about the API.

Parameters:

  • title (str) – the title of the application

  • version (str) – the version of the OpenRPC document

  • description (Union[UnsetType, str]) – a verbose description of the application

  • contact (Union[UnsetType, Contact]) – the contact information for the exposed API

  • license (Union[UnsetType, License]) – the license information for the exposed API

  • termsOfService (Union[UnsetType, str]) – a URL to the Terms of Service for the API

class pjrpc.server.specs.openrpc.ServerVariable(default, enum=UNSET, description=UNSET)[source]#

Bases: object

An object representing a Server Variable for server URL template substitution.

Parameters:

  • default (str) – The default value to use for substitution.

  • enum (Union[UnsetType, List[str]]) – An enumeration of string values to be used if the substitution options are from a limited set.

  • description (Union[UnsetType, str]) – An optional description for the server variable.

class pjrpc.server.specs.openrpc.Server(name, url, summary=UNSET, description=UNSET, variables=UNSET)[source]#

Bases: object

Connectivity information of a target server.

Parameters:

  • name (str) – a name to be used as the canonical name for the server.

  • url (str) – a URL to the target host. This URL supports Server Variables.

  • summary (Union[UnsetType, str]) – a short summary of what the server is

  • description (Union[UnsetType, str]) – an optional string describing the host designated by the URL

  • variables (Union[UnsetType, Dict[str, ServerVariable]]) – A map between a variable name and its value. The value is passed into the Runtime Expression to produce a server URL.

class pjrpc.server.specs.openrpc.ExternalDocumentation(url, description=UNSET)[source]#

Bases: object

Allows referencing an external resource for extended documentation.

Parameters:

  • url (str) – A verbose explanation of the target documentation

  • description (Union[UnsetType, str]) – The URL for the target documentation. Value MUST be in the format of a URL

class pjrpc.server.specs.openrpc.Tag(name, description=UNSET, externalDocs=UNSET)[source]#

Bases: object

A list of tags for API documentation control. Tags can be used for logical grouping of methods by resources or any other qualifier.

Parameters:
class pjrpc.server.specs.openrpc.ExampleObject(value, name, summary=UNSET, description=UNSET, externalValue=UNSET)[source]#

Bases: object

The ExampleObject object is an object the defines an example.

Parameters:

  • value (Any) – embedded literal example

  • name (str) – canonical name of the example

  • summary (Union[UnsetType, str]) – short description for the example

  • description (Union[UnsetType, str]) – a verbose explanation of the example

  • externalValue (Union[UnsetType, str]) – a URL that points to the literal example.

class pjrpc.server.specs.openrpc.MethodExample(name, params, result, summary=UNSET, description=UNSET)[source]#

Bases: object

The example Pairing object consists of a set of example params and result.

Parameters:
class pjrpc.server.specs.openrpc.ContentDescriptor(name, schema, summary=UNSET, description=UNSET, required=UNSET, deprecated=UNSET)[source]#

Bases: object

Content Descriptors are objects that describe content. They are reusable ways of describing either parameters or result.

Parameters:
class pjrpc.server.specs.openrpc.Error(code, message, data=UNSET)[source]#

Bases: object

Defines an application level error.

Parameters:

  • code (int) – a Number that indicates the error type that occurred

  • message (str) – a String providing a short description of the error

  • data (Union[UnsetType, Any]) – a Primitive or Structured value that contains additional information about the error

class pjrpc.server.specs.openrpc.ParamStructure(value)[source]#

Bases: str, Enum

The expected format of the parameters.

Bases: object

Parameters:

  • name (str) – Canonical name of the link.

  • description (Union[UnsetType, str]) – A description of the link.

  • summary (Union[UnsetType, str]) – Short description for the link.

  • method (Union[UnsetType, str]) – The name of an existing, resolvable OpenRPC method, as defined with a unique method.

  • params (Union[UnsetType, Any]) – A map representing parameters to pass to a method as specified with method. The key is the parameter name to be used, whereas the value can be a constant or a runtime expression to be evaluated and passed to the linked method.

  • server (Union[UnsetType, Server]) – A server object to be used by the target method.

class pjrpc.server.specs.openrpc.MethodInfo(name, params, result, errors=UNSET, links=UNSET, paramStructure=UNSET, examples=UNSET, summary=UNSET, description=UNSET, tags=UNSET, deprecated=UNSET, externalDocs=UNSET, servers=UNSET)[source]#

Bases: object

Describes the interface for the given method name.

Parameters:
class pjrpc.server.specs.openrpc.Components(contentDescriptors=UNSET, schemas=UNSET, examples=UNSET, links=UNSET, errors=UNSET, examplePairingObjects=UNSET, tags=UNSET)[source]#

Bases: object

Set of reusable objects for different aspects of the OpenRPC.

Parameters:
class pjrpc.server.specs.openrpc.SpecRoot(info, components, methods=<factory>, servers=UNSET, externalDocs=UNSET, openrpc='1.3.2')[source]#

Bases: object

The root object of the OpenRPC document.

Parameters:
class pjrpc.server.specs.openrpc.OpenRpcMeta[source]#

Bases: TypedDict

pjrpc.server.specs.openrpc.annotate(params_schema=UNSET, result_schema=UNSET, errors=UNSET, examples=UNSET, summary=UNSET, description=UNSET, tags=UNSET, deprecated=UNSET, external_docs=UNSET, servers=UNSET)[source]#

Adds JSON-RPC method to the API specification.

Parameters:
Return type:

Callable[[Func], Func]

class pjrpc.server.specs.openrpc.OpenRPC(info, path='/openrpc.json', servers=UNSET, external_docs=UNSET, openrpc='1.3.2', schema_extractor=None)[source]#

Bases: Specification

OpenRPC Specification.

Parameters:
schema(path, methods_map={})[source]#

Returns specification schema.

Parameters:

  • path (str) – methods endpoint path

  • methods_map (Mapping[str, Iterable[Method]]) – methods map the specification is generated for. Each item is a mapping from a prefix to methods on which the methods will be served

Return type:

Dict[str, Any]