fastapi-jsonrpc 0.1.21

JSON-RPC server based on fastapi

Description

JSON-RPC server based on fastapi:

https://fastapi.tiangolo.com

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

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(..., example='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(app, port=5000, debug=True, access_log=False)

Go to:

http://127.0.0.1:5000/docs

FastAPI dependencies usage example

pip install uvicorn

import fastapi_jsonrpc as jsonrpc
from fastapi import Body, Header, Depends


# 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'


# database models

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


class Account:
    def __init__(self, account_id, owner_name, balance, currency):
        self.account_id = account_id
        self.owner_name = owner_name
        self.balance = balance
        self.currency = currency

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


# fake database

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

accounts = {
    '1.1': Account('1.1', 'user1', 100, 'USD'),
    '1.2': Account('1.2', 'user1', 200, 'EUR'),
    '2.1': Account('2.1', 'user2', 300, 'USD'),
}


def get_user_by_token(auth_token) -> User:
    try:
        return users[auth_token]
    except KeyError:
        raise AuthError()


def get_account_by_id(account_id) -> Account:
    try:
        return accounts[account_id]
    except KeyError:
        raise AccountNotFound()


# dependencies

def get_auth_user(
    # this will become the header-parameter of json-rpc method that uses this dependency
    auth_token: str = Header(
        ...,
        alias='user-auth-token',
    ),
) -> User:
    return get_user_by_token(auth_token)


def get_account(
    # this will become the parameter of the json-rpc method that uses this dependency
    account_id: str,
    user: User = Depends(get_auth_user),
) -> Account:
    account = get_account_by_id(account_id)
    if not account.owned_by(user):
        raise AccountNotFound()
    return account


# 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,
    # 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),
) -> str:
    return f'{account.balance} {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),
) -> str:
    if account.balance - amount < 0:
        raise NotEnoughMoney
    account.balance -= amount
    return get_balance(account)


# JSON-RPC API

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


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

Go to:

http://127.0.0.1:5000/docs

Development

• Install poetry

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

• Install dephell

  pip install dephell

• Install dependencies

  poetry update

• Change dependencies

  Edit pyproject.toml

  poetry update
  dephell deps convert

• Bump version

  poetry version
  dephell deps convert

• Publish to pypi

  poetry publish --build