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
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(..., 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('example1:app', port=5000, debug=True, access_log=False)
Go to:
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)
Go to:
http://127.0.0.1:5000/docs
Development
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.tomlpoetry update dephell deps convert
Bump version
poetry version patch dephell deps convert
Publish to pypi
poetry publish --build
122 Dec 27, 2022
1 Oct 31, 2021
8 Feb 8, 2022
95 Jan 8, 2023
71 Dec 30, 2022
28 Oct 25, 2022
381 Dec 11, 2022
131 Jan 1, 2023
54 Dec 13, 2022
18 Oct 31, 2022
139 Dec 27, 2022
78 Jan 3, 2023
417 Jan 7, 2023
371 Jan 1, 2023
83 Dec 19, 2022
59 Dec 11, 2022
268 Jan 1, 2023
315 Dec 25, 2022