fastapi-jsonrpc 0.1.21

说明

基于FastAPI的JSON-RPC服务器：

https://fastapi.tiangolo.com

安装

pip install fastapi-jsonrpc

文档

阅读FastAPI文档并参见下面的使用示例

简单用法示例

pip install uvicorn

importfastapi_jsonrpcasjsonrpcfrompydanticimportBaseModelfromfastapiimportBodyapp=jsonrpc.API()api_v1=jsonrpc.Entrypoint('/api/v1/jsonrpc')classMyError(jsonrpc.BaseError):CODE=5000MESSAGE='My error'classDataModel(BaseModel):details:str@api_v1.method(errors=[MyError])defecho(data:str=Body(...,example='123'),)->str:ifdata=='error':raiseMyError(data={'details':'error'})else:returndataapp.bind_entrypoint(api_v1)if__name__=='__main__':importuvicornuvicorn.run(app,port=5000,debug=True,access_log=False)

转到：

http://127.0.0.1:5000/docs

FastAPI依赖项使用示例

pip install uvicorn

importfastapi_jsonrpcasjsonrpcfromfastapiimportBody,Header,Depends# errorsclassAuthError(jsonrpc.BaseError):CODE=7000MESSAGE='Auth error'classAccountNotFound(jsonrpc.BaseError):CODE=6000MESSAGE='Account not found'classNotEnoughMoney(jsonrpc.BaseError):CODE=6001MESSAGE='Not enough money'# database modelsclassUser:def__init__(self,name):self.name=nameclassAccount:def__init__(self,account_id,owner_name,balance,currency):self.account_id=account_idself.owner_name=owner_nameself.balance=balanceself.currency=currencydefowned_by(self,user:User):returnself.owner_name==user.name# fake databaseusers={'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'),}defget_user_by_token(auth_token)->User:try:returnusers[auth_token]exceptKeyError:raiseAuthError()defget_account_by_id(account_id)->Account:try:returnaccounts[account_id]exceptKeyError:raiseAccountNotFound()# dependenciesdefget_auth_user(# this will become the header-parameter of json-rpc method that uses this dependencyauth_token:str=Header(...,alias='user-auth-token',),)->User:returnget_user_by_token(auth_token)defget_account(# this will become the parameter of the json-rpc method that uses this dependencyaccount_id:str,user:User=Depends(get_auth_user),)->Account:account=get_account_by_id(account_id)ifnotaccount.owned_by(user):raiseAccountNotFound()returnaccount# JSON-RPC entrypointcommon_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 requestdependencies=[Depends(get_auth_user)],# this dependencies called separately for every json-rpc request in batch requestcommon_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()defget_balance(account:Account=Depends(get_account),)->str:returnf'{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])defwithdraw(account:Account=Depends(get_account),amount:int=Body(...,gt=0),)->str:ifaccount.balance-amount<0:raiseNotEnoughMoneyaccount.balance-=amountreturnget_balance(account)# JSON-RPC APIapp=jsonrpc.API()app.bind_entrypoint(api_v1)if__name__=='__main__':importuvicornuvicorn.run(app,port=5000,debug=True,access_log=False)

转到：

http://127.0.0.1:5000/docs

开发

• 安装诗歌

  https://github.com/sdispater/poetry#installation

• 安装deshell

  pip install dephell
  

• 安装依赖项

  poetry update
  

• 更改依赖项

  Edit ^{tt1}$

  poetry update
  dephell deps convert
  

• 凹凸版本

  poetry version
  dephell deps convert
  

• 发布到pypi

  poetry publish --build