A modern `Json-Rpc/Bson-Rpc` implementation, compatible with `Json-Rpc Ast` and `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 Ast
and 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
-
自定义
-
- 多种Runtime支持
- 兼容 Json-Rpc Ast 规范
- 兼容 Json-Rpc 2.0 规范
- 兼容 Json-Rpc X 规范
- 支持完备的链式调用
- 支持嵌套调用
- 支持双向调用
- 双向流式传输
- 支持 BSON
- IDE 支持
- 分布式Server
使用
入门
Server 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.network.server.aioserver import AioServer
namespace = {
"ping": lambda x: f"pong {x}",
}
rpc = JsonRpc(name = "server_test", namespace=namespace)
async def main():
await rpc.run_server(AioServer(path="/func"))
asyncio.run(main())
Client 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
rpc = JsonRpc(
name="client_test",
)
async def main():
client = AioClient(path="/func")
await rpc.run_client(client)
ping = ProxyObj(rpc).ping
assert await ping("hello") == "pong hello"
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
链式调用
Server 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
rpc = JsonRpc(
name="client_test",
)
class Test:
def __init__(self, a, b) -> None:
self.a = a
self.b = b
def output(self, name):
return f"hello: {name}, a: {self.a}, b: {self.b}"
namespace = {
"Test": Test
}
rpc = JsonRpc(name = "server_test", namespace=namespace)
async def main():
await rpc.run_server(AioServer(path="/chain"))
asyncio.run(main())
Client 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
rpc = JsonRpc(
name="client_test",
)
async def main():
client = AioClient(path="/func")
await rpc.run_client(client)
ping = ProxyObj(rpc).ping
assert await ping("hello") == "pong hello"
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
反向调用
Server 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer
namespace = {
"ping": lambda x: f"pong {x}",
}
rpc = JsonRpc(name = "server_test", namespace=namespace)
async def reverse_call(name: str):
res = await ProxyObj(rpc, name=name).ping(name)
return res
rpc.add_namespace({
"reverse_call": reverse_call,
})
async def main():
await rpc.run_server(AioServer(path="/ws"))
asyncio.run(main())
Client 端
from pprint import pprint
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
namespace = {
"ping": lambda x: f"pong {x}",
}
rpc = JsonRpc(
name="client_test",
namespace=namespace
)
async def main():
client = AioClient(path="/ws")
await rpc.run_client(client)
proxy1 = ProxyObj(rpc).reverse_call
assert await proxy1("hello") == "pong hello"
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
流式调用
Server 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer
async def test_async_iter(n: int):
for i in range(n):
await asyncio.sleep(1)
yield i
namespace = {
"test_async_iter": test_async_iter,
}
rpc = JsonRpc(name = "server_test", namespace=namespace)
async def main():
await rpc.run_server(AioServer(path="/ws"))
asyncio.run(main())
Client 端
from pprint import pprint
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
namespace = {
"ping": lambda x: f"pong {x}",
}
rpc = JsonRpc(
name="client_test",
namespace=namespace
)
async def main():
client = AioClient(path="/ws")
await rpc.run_client(client)
proxy3 = ProxyObj(rpc).test_async_iter
async for i in proxy3(3):
print(i)
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
嵌套调用
Server 端
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer
class Test:
def __init__(self, a, b) -> None:
self.a = a
self.b = b
def output(self, name):
return f"hello: {name}"
def class_sum(obj: Test):
return obj.a + obj.b
namespace = {
"class_sum": class_sum,
"Test": Test,
}
rpc = JsonRpc(name = "server_test", namespace=namespace)
async def main():
await rpc.run_server(AioServer(path="/ws"))
asyncio.run(main())
Client 端
from pprint import pprint
import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient
namespace = {
"ping": lambda x: f"pong {x}",
}
rpc = JsonRpc(
name="client_test",
namespace=namespace
)
async def main():
client = AioClient(path="/ws")
await rpc.run_client(client)
proxy1 = ProxyObj(rpc).Test(1, b=2)
proxy2 = ProxyObj(rpc).class_sum(proxy1)
assert await proxy2 == 3
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
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.2.0b6.tar.gz
(13.5 kB
view hashes)
Built Distribution
lacia-0.2.0b6-py3-none-any.whl
(17.8 kB
view hashes)