JSON-RPC server based on fastapi
Project description
Description
JSON-RPC server based on fastapi:
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:
FastAPI dependencies usage example
pip install uvicorn
example2.py
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('example2:app', port=5000, debug=True, access_log=False)
Go to:
Development
Install poetry
Install dephell
pip install dephell
Install dependencies
poetry update
Regenerate README.rst
rst_include include -s README.src.rst -t README.rst -q
Change dependencies
Edit pyproject.toml
poetry update dephell deps convert
Bump version
poetry version dephell deps convert
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-0.1.24.tar.gz
(11.8 kB
view hashes)
Built Distribution
Close
Hashes for fastapi_jsonrpc-0.1.24-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4c3f96e7cc4fb97be121b33c4de1f89458ebc7557be6b1a51da1a6704e04ea9d |
|
MD5 | 456438c96f1335081edea23d18b2b9b1 |
|
BLAKE2b-256 | 0822828fb558fac0b70e958a088959b1707462046628acbda8cb72695d17124d |