A modern `Json-Rpc/Bson-Rpc` implementation, compatible with `Json-Rpc 2.0` and `Json-Rpc X`, supports multiple network protocols and backend frameworks and supports bidirectional calls.
Project description
Lacia
A modern Json-Rpc/Bson-Rpc
implementation, compatible with Json-Rpc 2.0
and Json-Rpc X
, supports multiple network protocols and backend frameworks and supports bidirectional calls..
人间总有一两风,填我十万八千梦
安装
pip install lacia
pdm add lacia
特性
- 跨语言
-
Python3
-
JavaScript
-
Go
-
C/C++
-
Java
-
Rust
-
- 多种网络协议支持
-
HTTP
-
WebSocket
-
- 兼容 Json-Rpc 2.0 规范
- 兼容 Json-Rpc X 规范
- 支持完备的链式调用
- 自动 Json-Rpc 规范转换
- 双向流式传输
- 支持 BSON
- IDE 支持 (#1)
- 分布式Server
使用
入门
Server 端
import asyncio
from lacia import JsonRpc, AioServer
expose = {
'add': lambda a, b: a + b,
'sub': lambda a, b: a - b,
'value': 'jsonrpc Server 0.0.1'
}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
rpc.run_server(AioServer())
Client 端
import asyncio
from lacia import JsonRpc, AioClient, logger
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', loop=loop)
async def main():
await rpc.run_client(AioClient())
res1 = await rpc.value
res2 = await rpc.add(4, 4)
logger.info(res1)
logger.info(res2)
loop.run_until_complete(main())
loop.run_forever()
链式调用
Server 端
import asyncio
from lacia import JsonRpc, AioServer
class number:
def __init__(self) -> None:
self.value = 0
def add(self, i: int):
self.value += i
return self
def sub(self, i: int):
self.value -= i
return self
expose = {'number': number}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
rpc.run_server(AioServer())
Client 端
import asyncio
from lacia import JsonRpc, AioClient, logger
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', loop=loop)
async def main():
await rpc.run_client(AioClient())
res3 = await rpc.number().add(10).sub(10).value
logger.info(res3)
loop.run_until_complete(main())
loop.run_forever()
反向调用
Server 端
import asyncio
from lacia import JsonRpc, AioServer
expose = {'value': 'jsonrpc Server 0.0.1'}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
async def repeat():
res = await rpc.value
return res
rpc.add_namespace('repeat', repeat)
rpc.run_server(AioServer())
Client 端
import asyncio
from lacia import JsonRpc, AioClient, logger
expose = {'value': 'jsonrpc Client 0.0.1'}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
async def main():
await rpc.run_client(AioClient())
res4 = await rpc.repeat()
logger.info(res4)
loop.run_until_complete(main())
loop.run_forever()
流式调用
Server 端
import asyncio
from lacia import JsonRpc, AioServer
async def async_generator(i = 10):
for i in range(i):
await asyncio.sleep(0.5)
yield i
expose = {'async_generator': async_generator}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
rpc.run_server(AioServer())
Client 端
import asyncio
from lacia import JsonRpc, AioClient, logger
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', loop=loop)
async def main():
await rpc.run_client(AioClient())
async for i in rpc.async_generator(5): # type: ignore
logger.info(i)
loop.run_until_complete(main())
loop.run_forever()
IDE 支持(实验性)
Server 端
import asyncio
from lacia import JsonRpc, AioServer
async def async_generator(i: int = 10):
for i in range(i):
await asyncio.sleep(0.5)
yield i
expose = {'async_generator': async_generator}
loop = asyncio.new_event_loop()
rpc = JsonRpc('/test', namespace=expose, loop=loop)
rpc.rpc.generate_pyi('CustomSchema') # Generate Pyi file in ./laciaschema/CustomSchema.pyi
rpc.run_server(AioServer())
Client 端
import asyncio
from lacia import JsonRpc, AioClient, logger
from laciaschema.CustomSchema import CustomSchema
loop = asyncio.new_event_loop()
irpc = JsonRpc('/test', loop=loop)
rpc = irpc.with_schema(CustomSchema) # Use CustomSchema as schema and
async def main():
await irpc.run_client(AioClient())
async for i in rpc.async_generator(5): # type: ignore
logger.info(i)
loop.run_until_complete(main())
loop.run_forever()
IDE 现只支持函数和基础类型, Class 正在重构中
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
lacia-0.0.5.tar.gz
(14.1 kB
view hashes)
Built Distribution
lacia-0.0.5-py3-none-any.whl
(19.5 kB
view hashes)