Skip to main content

JSON-RPC server based on fastapi

Project description

tests

Description

JSON-RPC server based on fastapi:

https://fastapi.tiangolo.com

OpenRPC supported too.

Motivation

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

Installation

pip install fastapi-jsonrpc

Documentation

Read FastAPI documentation and see usage examples bellow

Simple usage example

pip install uvicorn

example1.py

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


@api_v1.method(errors=[MyError])
def echo(
    data: str = Body(..., examples=['123']),
) -> str:
    if data == 'error':
        raise MyError(data={'details': 'error'})
    else:
        return data


app.bind_entrypoint(api_v1)


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

OpenRPC:

http://127.0.0.1:5000/openrpc.json

Swagger:

http://127.0.0.1:5000/docs

FastAPI dependencies usage example

pip install uvicorn

example2.py

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):
        self.name = name

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


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(
        None,
        alias='user-auth-token',
    ),
) -> User:
    if not auth_token:
        raise AuthError

    try:
        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:
    try:
        account = get_account_by_id(account_id)
    except KeyError:
        raise AccountNotFound

    if not account.owned_by(user):
        raise AccountNotFound

    return account


# JSON-RPC middlewares

@asynccontextmanager
async def logging_middleware(ctx: jsonrpc.JsonRpcContext):
    logger.info('Request: %r', ctx.raw_request)
    try:
        yield
    finally:
        logger.info('Response: %r', ctx.raw_response)


# JSON-RPC entrypoint

common_errors = [AccountNotFound, AuthError]
common_errors.extend(jsonrpc.Entrypoint.default_errors)

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'
    '/api/v1/jsonrpc',
    errors=common_errors,
    middlewares=[logging_middleware],
    # this dependencies called once for whole json-rpc batch request
    dependencies=[Depends(get_auth_user)],
    # this dependencies called separately for every json-rpc request in batch request
    common_dependencies=[Depends(get_account)],
)


# JSON-RPC methods of this entrypoint

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


# this json-rpc method has two json-rpc-parameters 'account_id', 'amount' and one header parameter 'user-auth-token'
@api_v1.method(errors=[NotEnoughMoney])
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)


# JSON-RPC API

app = jsonrpc.API()
app.bind_entrypoint(api_v1)


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

OpenRPC:

http://127.0.0.1:5000/openrpc.json

Swagger:

http://127.0.0.1:5000/docs

./images/fastapi-jsonrpc.png

Development

  • Install poetry

    https://github.com/sdispater/poetry#installation

  • 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


Release history Release notifications | RSS feed

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-3.1.2.tar.gz (16.2 kB view details)

Uploaded Source

Built Distribution

fastapi_jsonrpc-3.1.2-py3-none-any.whl (15.1 kB view details)

Uploaded Python 3

File details

Details for the file fastapi_jsonrpc-3.1.2.tar.gz.

File metadata

  • Download URL: fastapi_jsonrpc-3.1.2.tar.gz
  • Upload date:
  • Size: 16.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.6.1 CPython/3.11.5 Darwin/23.6.0

File hashes

Hashes for fastapi_jsonrpc-3.1.2.tar.gz
Algorithm Hash digest
SHA256 51ff69c7e5c645d18b27d031e54290b0e44e8d62676bbe05bbfd6330dd5ef282
MD5 4ad94255271d2912e20af90b80f0cc8f
BLAKE2b-256 5c4ae765442e4109a512d7445d4416027ca698778c0f2cb9d34d166c68fa1746

See more details on using hashes here.

File details

Details for the file fastapi_jsonrpc-3.1.2-py3-none-any.whl.

File metadata

  • Download URL: fastapi_jsonrpc-3.1.2-py3-none-any.whl
  • Upload date:
  • Size: 15.1 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: poetry/1.6.1 CPython/3.11.5 Darwin/23.6.0

File hashes

Hashes for fastapi_jsonrpc-3.1.2-py3-none-any.whl
Algorithm Hash digest
SHA256 a06e3afae6df5c8a185c9569b8dcec45113becd7c0ccaa3ab7aec1ee954aea3a
MD5 d30a07bc17d0614874a9f25bcfa54097
BLAKE2b-256 a9208e9a4de8355d866bc07c81512d6b506387cc16ac5f9b81eb231c8776d436

See more details on using hashes here.

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