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
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:
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:
Development
Install poetry
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
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.23.tar.gz
(11.6 kB
view hashes)
Built Distribution
Close
Hashes for fastapi_jsonrpc-0.1.23-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 86734db1eab5b1325bf637e1b681d665e7a0e2169b5208c9f75767c113c70ae6 |
|
MD5 | f7839db975b75d3ff7a3c93edab53664 |
|
BLAKE2b-256 | bae4bb508556eac10d8238db518b9901f1da54ad856fe2c50196c85c6f219f2f |