Awesome Open Source
Awesome Open Source



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 dephell

    pip install dephell
  • Install dependencies

    poetry update
  • Regenerate README.rst

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

    Edit pyproject.toml

    poetry update
    dephell deps convert
  • Bump version

    poetry version patch
    dephell deps convert
  • Publish to pypi

    poetry publish --build

Get A Weekly Email With Trending Projects For These Topics
No Spam. Unsubscribe easily at any time.
Python (1,124,651
Swagger (2,572
Openapi (1,376
Fastapi (872
Json Rpc (395
Asgi (167
Starlette (144
Pydantic (128
Related Projects