Skip to main content

JSON-RPC server based on fastapi

Project description



JSON-RPC server based on fastapi:


Autogenerated OpenAPI and Swagger (thanks to fastapi) for JSON-RPC!!!


pip install fastapi-jsonrpc


Read FastAPI documentation and see usage examples bellow

Simple usage example

pip install uvicorn

import fastapi_jsonrpc as jsonrpc
from pydantic import BaseModel
from fastapi import Body

app = jsonrpc.API()

api_v1 = jsonrpc.Entrypoint('/api/v1/jsonrpc')

class MyError(jsonrpc.BaseError):
    CODE = 5000
    MESSAGE = 'My error'

    class DataModel(BaseModel):
        details: str

def echo(
    data: str = Body(..., example='123'),
) -> str:
    if data == 'error':
        raise MyError(data={'details': 'error'})
        return data


if __name__ == '__main__':
    import uvicorn'example1:app', port=5000, debug=True, access_log=False)

Go to:

FastAPI dependencies usage example

pip install uvicorn

import logging
from contextlib import asynccontextmanager

from pydantic import BaseModel, Field
import fastapi_jsonrpc as jsonrpc
from fastapi import Body, Header, Depends

logger = logging.getLogger(__name__)

# database models

class User:
    def __init__(self, name): = name

    def __eq__(self, other):
        if not isinstance(other, User):
            return False
        return ==

class Account:
    def __init__(self, account_id, owner, amount, currency):
        self.account_id = account_id
        self.owner = owner
        self.amount = amount
        self.currency = currency

    def owned_by(self, user: User):
        return self.owner == user

# fake database

users = {
    '1': User('user1'),
    '2': User('user2'),

accounts = {
    '1.1': Account('1.1', users['1'], 100, 'USD'),
    '1.2': Account('1.2', users['1'], 200, 'EUR'),
    '2.1': Account('2.1', users['2'], 300, 'USD'),

def get_user_by_token(auth_token) -> User:
    return users[auth_token]

def get_account_by_id(account_id) -> Account:
    return accounts[account_id]

# schemas

class Balance(BaseModel):
    """Account balance"""
    amount: int = Field(..., example=100)
    currency: str = Field(..., example='USD')

# errors

class AuthError(jsonrpc.BaseError):
    CODE = 7000
    MESSAGE = 'Auth error'

class AccountNotFound(jsonrpc.BaseError):
    CODE = 6000
    MESSAGE = 'Account not found'

class NotEnoughMoney(jsonrpc.BaseError):
    CODE = 6001
    MESSAGE = 'Not enough money'

    class DataModel(BaseModel):
        balance: Balance

# dependencies

def get_auth_user(
    # this will become the header-parameter of json-rpc method that uses this dependency
    auth_token: str = Header(
) -> User:
    if not auth_token:
        raise AuthError

        return get_user_by_token(auth_token)
    except KeyError:
        raise AuthError

def get_account(
    # this will become the parameter of the json-rpc method that uses this dependency
    account_id: str = Body(..., example='1.1'),
    user: User = Depends(get_auth_user),
) -> Account:
        account = get_account_by_id(account_id)
    except KeyError:
        raise AccountNotFound

    if not account.owned_by(user):
        raise AccountNotFound

    return account

# JSON-RPC middlewares

async def logging_middleware(ctx: jsonrpc.JsonRpcContext):'Request: %r', ctx.raw_request)
    finally:'Response: %r', ctx.raw_response)

# JSON-RPC entrypoint

common_errors = [AccountNotFound, AuthError]

api_v1 = jsonrpc.Entrypoint(
    # Swagger shows for entrypoint common parameters gathered by dependencies and common_dependencies:
    #    - json-rpc-parameter 'account_id'
    #    - header parameter 'user-auth-token'
    # this dependencies called once for whole json-rpc batch request
    # this dependencies called separately for every json-rpc request in batch request

# JSON-RPC methods of this entrypoint

# this json-rpc method has one json-rpc-parameter 'account_id' and one header parameter 'user-auth-token'
def get_balance(
    account: Account = Depends(get_account),
) -> Balance:
    return Balance(

# this json-rpc method has two json-rpc-parameters 'account_id', 'amount' and one header parameter 'user-auth-token'
def withdraw(
    account: Account = Depends(get_account),
    amount: int = Body(..., gt=0, example=10),
) -> Balance:
    if account.amount - amount < 0:
        raise NotEnoughMoney(data={'balance': get_balance(account)})
    account.amount -= amount
    return get_balance(account)


app = jsonrpc.API()

if __name__ == '__main__':
    import uvicorn'example2:app', port=5000, debug=True, access_log=False)

Go to:



  • Install poetry

  • Install dependencies

    poetry update
  • Regenerate README.rst

    rst_include include -q README.src.rst README.rst
  • Change dependencies

    Edit pyproject.toml

    poetry update
  • Bump version

    poetry version patch
    poetry version minor
    poetry version major
  • Publish to pypi

    poetry publish --build

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fastapi-jsonrpc-2.4.1.tar.gz (15.3 kB view hashes)

Uploaded source

Built Distribution

fastapi_jsonrpc-2.4.1-py3-none-any.whl (13.0 kB view hashes)

Uploaded py3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page