|tests|
.. |tests| image:: https://github.com/smagafurov/fastapi-jsonrpc/actions/workflows/tests.yml/badge.svg
:target: https://github.com/smagafurov/fastapi-jsonrpc/actions/workflows/tests.yml
Description
===========
JSON-RPC server based on fastapi:
https://fastapi.tiangolo.com
OpenRPC supported too.
Motivation
^^^^^^^^^^
Autogenerated **OpenAPI** and **Swagger** (thanks to fastapi) for JSON-RPC!!!
Installation
============
.. code-block:: bash
pip install fastapi-jsonrpc
Documentation
=============
Read FastAPI documentation and see usage examples bellow
Simple usage example
====================
.. code-block:: bash
pip install uvicorn
example1.py
.. code-block:: python
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(..., examples=['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)
OpenRPC:
http://127.0.0.1:5000/openrpc.json
Swagger:
http://127.0.0.1:5000/docs
FastAPI dependencies usage example
==================================
.. code-block:: bash
pip install uvicorn
example2.py
.. code-block:: python
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)
OpenRPC:
http://127.0.0.1:5000/openrpc.json
Swagger:
http://127.0.0.1:5000/docs
.. image:: ./images/fastapi-jsonrpc.png
Development
===========
* Install poetry
https://github.com/sdispater/poetry#installation
* Install dependencies
.. code-block:: bash
poetry update
* Regenerate README.rst
.. code-block:: bash
rst_include include -q README.src.rst README.rst
* Change dependencies
Edit ``pyproject.toml``
.. code-block:: bash
poetry update
* Bump version
.. code-block:: bash
poetry version patch
poetry version minor
poetry version major
* Publish to pypi
.. code-block:: bash
poetry publish --build
Raw data
{
"_id": null,
"home_page": "https://github.com/smagafurov/fastapi-jsonrpc",
"name": "fastapi-jsonrpc",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "json-rpc, asgi, swagger, openapi, fastapi, pydantic, starlette",
"author": "Sergey Magafurov",
"author_email": "magafurov@tochka.com",
"download_url": "https://files.pythonhosted.org/packages/d6/78/386045a812d901c358300fcd4e37b86b74a38a9db1cd9e48cd8d24d331be/fastapi_jsonrpc-3.2.1.tar.gz",
"platform": null,
"description": "|tests|\n\n.. |tests| image:: https://github.com/smagafurov/fastapi-jsonrpc/actions/workflows/tests.yml/badge.svg\n :target: https://github.com/smagafurov/fastapi-jsonrpc/actions/workflows/tests.yml\n\n\nDescription\n===========\n\nJSON-RPC server based on fastapi:\n\n https://fastapi.tiangolo.com\n\nOpenRPC supported too.\n\nMotivation\n^^^^^^^^^^\n\nAutogenerated **OpenAPI** and **Swagger** (thanks to fastapi) for JSON-RPC!!!\n\nInstallation\n============\n\n.. code-block:: bash\n\n pip install fastapi-jsonrpc\n\nDocumentation\n=============\n\nRead FastAPI documentation and see usage examples bellow\n\nSimple usage example\n====================\n\n.. code-block:: bash\n\n pip install uvicorn\n\nexample1.py\n\n.. code-block:: python\n\n import fastapi_jsonrpc as jsonrpc\n from pydantic import BaseModel\n from fastapi import Body\n\n\n app = jsonrpc.API()\n\n api_v1 = jsonrpc.Entrypoint('/api/v1/jsonrpc')\n\n\n class MyError(jsonrpc.BaseError):\n CODE = 5000\n MESSAGE = 'My error'\n\n class DataModel(BaseModel):\n details: str\n\n\n @api_v1.method(errors=[MyError])\n def echo(\n data: str = Body(..., examples=['123']),\n ) -> str:\n if data == 'error':\n raise MyError(data={'details': 'error'})\n else:\n return data\n\n\n app.bind_entrypoint(api_v1)\n\n\n if __name__ == '__main__':\n import uvicorn\n uvicorn.run('example1:app', port=5000, debug=True, access_log=False)\n\nOpenRPC:\n\n http://127.0.0.1:5000/openrpc.json\n\nSwagger:\n\n http://127.0.0.1:5000/docs\n\nFastAPI dependencies usage example\n==================================\n\n.. code-block:: bash\n\n pip install uvicorn\n\nexample2.py\n\n.. code-block:: python\n\n import logging\n from contextlib import asynccontextmanager\n\n from pydantic import BaseModel, Field\n import fastapi_jsonrpc as jsonrpc\n from fastapi import Body, Header, Depends\n\n\n logger = logging.getLogger(__name__)\n\n\n # database models\n\n class User:\n def __init__(self, name):\n self.name = name\n\n def __eq__(self, other):\n if not isinstance(other, User):\n return False\n return self.name == other.name\n\n\n class Account:\n def __init__(self, account_id, owner, amount, currency):\n self.account_id = account_id\n self.owner = owner\n self.amount = amount\n self.currency = currency\n\n def owned_by(self, user: User):\n return self.owner == user\n\n\n # fake database\n\n users = {\n '1': User('user1'),\n '2': User('user2'),\n }\n\n accounts = {\n '1.1': Account('1.1', users['1'], 100, 'USD'),\n '1.2': Account('1.2', users['1'], 200, 'EUR'),\n '2.1': Account('2.1', users['2'], 300, 'USD'),\n }\n\n\n def get_user_by_token(auth_token) -> User:\n return users[auth_token]\n\n\n def get_account_by_id(account_id) -> Account:\n return accounts[account_id]\n\n\n # schemas\n\n class Balance(BaseModel):\n \"\"\"Account balance\"\"\"\n amount: int = Field(..., example=100)\n currency: str = Field(..., example='USD')\n\n\n # errors\n\n class AuthError(jsonrpc.BaseError):\n CODE = 7000\n MESSAGE = 'Auth error'\n\n\n class AccountNotFound(jsonrpc.BaseError):\n CODE = 6000\n MESSAGE = 'Account not found'\n\n\n class NotEnoughMoney(jsonrpc.BaseError):\n CODE = 6001\n MESSAGE = 'Not enough money'\n\n class DataModel(BaseModel):\n balance: Balance\n\n\n # dependencies\n\n def get_auth_user(\n # this will become the header-parameter of json-rpc method that uses this dependency\n auth_token: str = Header(\n None,\n alias='user-auth-token',\n ),\n ) -> User:\n if not auth_token:\n raise AuthError\n\n try:\n return get_user_by_token(auth_token)\n except KeyError:\n raise AuthError\n\n\n def get_account(\n # this will become the parameter of the json-rpc method that uses this dependency\n account_id: str = Body(..., example='1.1'),\n user: User = Depends(get_auth_user),\n ) -> Account:\n try:\n account = get_account_by_id(account_id)\n except KeyError:\n raise AccountNotFound\n\n if not account.owned_by(user):\n raise AccountNotFound\n\n return account\n\n\n # JSON-RPC middlewares\n\n @asynccontextmanager\n async def logging_middleware(ctx: jsonrpc.JsonRpcContext):\n logger.info('Request: %r', ctx.raw_request)\n try:\n yield\n finally:\n logger.info('Response: %r', ctx.raw_response)\n\n\n # JSON-RPC entrypoint\n\n common_errors = [AccountNotFound, AuthError]\n common_errors.extend(jsonrpc.Entrypoint.default_errors)\n\n api_v1 = jsonrpc.Entrypoint(\n # Swagger shows for entrypoint common parameters gathered by dependencies and common_dependencies:\n # - json-rpc-parameter 'account_id'\n # - header parameter 'user-auth-token'\n '/api/v1/jsonrpc',\n errors=common_errors,\n middlewares=[logging_middleware],\n # this dependencies called once for whole json-rpc batch request\n dependencies=[Depends(get_auth_user)],\n # this dependencies called separately for every json-rpc request in batch request\n common_dependencies=[Depends(get_account)],\n )\n\n\n # JSON-RPC methods of this entrypoint\n\n # this json-rpc method has one json-rpc-parameter 'account_id' and one header parameter 'user-auth-token'\n @api_v1.method()\n def get_balance(\n account: Account = Depends(get_account),\n ) -> Balance:\n return Balance(\n amount=account.amount,\n currency=account.currency,\n )\n\n\n # this json-rpc method has two json-rpc-parameters 'account_id', 'amount' and one header parameter 'user-auth-token'\n @api_v1.method(errors=[NotEnoughMoney])\n def withdraw(\n account: Account = Depends(get_account),\n amount: int = Body(..., gt=0, example=10),\n ) -> Balance:\n if account.amount - amount < 0:\n raise NotEnoughMoney(data={'balance': get_balance(account)})\n account.amount -= amount\n return get_balance(account)\n\n\n # JSON-RPC API\n\n app = jsonrpc.API()\n app.bind_entrypoint(api_v1)\n\n\n if __name__ == '__main__':\n import uvicorn\n uvicorn.run('example2:app', port=5000, debug=True, access_log=False)\n\nOpenRPC:\n\n http://127.0.0.1:5000/openrpc.json\n\nSwagger:\n\n http://127.0.0.1:5000/docs\n\n.. image:: ./images/fastapi-jsonrpc.png\n\nDevelopment\n===========\n\n* Install poetry\n\n https://github.com/sdispater/poetry#installation\n\n* Install dependencies\n\n .. code-block:: bash\n\n poetry update\n\n* Regenerate README.rst\n\n .. code-block:: bash\n\n rst_include include -q README.src.rst README.rst\n\n* Change dependencies\n\n Edit ``pyproject.toml``\n\n .. code-block:: bash\n\n poetry update\n\n* Bump version\n\n .. code-block:: bash\n\n poetry version patch\n poetry version minor\n poetry version major\n\n* Publish to pypi\n\n .. code-block:: bash\n\n poetry publish --build\n\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "JSON-RPC server based on fastapi",
"version": "3.2.1",
"project_urls": {
"Homepage": "https://github.com/smagafurov/fastapi-jsonrpc",
"Repository": "https://github.com/smagafurov/fastapi-jsonrpc"
},
"split_keywords": [
"json-rpc",
" asgi",
" swagger",
" openapi",
" fastapi",
" pydantic",
" starlette"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "5ca3f5b8730e466e69290cc59201b89b1ecb233a2e6b1d6b0b77e56602ac02a3",
"md5": "45a70bffbf3d802451dce354a45a8e35",
"sha256": "52f5a1c16fa0eefa0fe7ae7dce4ea85611cf3ca77b206a4711ecaf26faaf02b3"
},
"downloads": -1,
"filename": "fastapi_jsonrpc-3.2.1-py3-none-any.whl",
"has_sig": false,
"md5_digest": "45a70bffbf3d802451dce354a45a8e35",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 15109,
"upload_time": "2024-09-24T19:51:31",
"upload_time_iso_8601": "2024-09-24T19:51:31.399247Z",
"url": "https://files.pythonhosted.org/packages/5c/a3/f5b8730e466e69290cc59201b89b1ecb233a2e6b1d6b0b77e56602ac02a3/fastapi_jsonrpc-3.2.1-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "d678386045a812d901c358300fcd4e37b86b74a38a9db1cd9e48cd8d24d331be",
"md5": "e832de07ce1153d4e2e3a27e91775c8f",
"sha256": "8b871e672e513fb7e74ab7ec5783d2ee84a7e744b21841b0c51faaa3d594a386"
},
"downloads": -1,
"filename": "fastapi_jsonrpc-3.2.1.tar.gz",
"has_sig": false,
"md5_digest": "e832de07ce1153d4e2e3a27e91775c8f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 16208,
"upload_time": "2024-09-24T19:51:32",
"upload_time_iso_8601": "2024-09-24T19:51:32.781702Z",
"url": "https://files.pythonhosted.org/packages/d6/78/386045a812d901c358300fcd4e37b86b74a38a9db1cd9e48cd8d24d331be/fastapi_jsonrpc-3.2.1.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-09-24 19:51:32",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "smagafurov",
"github_project": "fastapi-jsonrpc",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "fastapi-jsonrpc"
}
JSON
