JSON-RPC server based on fastapi
Project description
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 dependencies
poetry update
Regenerate README.rst
rst_include include -q README.src.rst README.rst
Change dependencies
Edit pyproject.toml
poetry update dephell deps convert
Bump version
poetry version patch 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-2.2.2.tar.gz
(15.4 kB
view hashes)
Built Distribution
Close
Hashes for fastapi_jsonrpc-2.2.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 3df7b959e3804c747f94a740e282135c7e082a646ab279a166a26734ee88cdfe |
|
MD5 | 8f0b4aa8d60bf7ffb53d2f500aae2aa9 |
|
BLAKE2-256 | c30393326464603ffbf2b1e4c5b269afa9ac1953f540e884a611aface8751399 |