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
341 Jan 03, 2023
66 Nov 09, 2022
33 Dec 15, 2022
2 Mar 14, 2022
329 Nov 10, 2022
15 Nov 23, 2022
91 Dec 31, 2022
39 Nov 24, 2022
106 Jan 06, 2023
0 Sep 20, 2022
417 Jan 07, 2023
139 Dec 27, 2022
31 Dec 09, 2022
10 Nov 30, 2022
28 Oct 25, 2022
2 Nov 07, 2021
95 Jan 08, 2023
2.2k Jan 08, 2023
122 Dec 27, 2022
1.3k Jan 04, 2023