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>=0.2

pdm add lacia>=0.2

特性

跨语言
- Python3
- JavaScript/TypeScript
- Golang
- C/C++
- Java
- Rust
多种网络协议支持
- HTTP
- WebSocket
- 自定义
多种Runtime支持
- 兼容 Json-Rpc Ast 规范
- 兼容 Json-Rpc 2.0 规范
- 兼容 Json-Rpc X 规范
支持完备的链式调用
支持嵌套调用
支持双向调用 (StoC, CtoS, CtoC)
双向流式传输
支持 BSON
IDE 支持
分布式Server

使用

入门

Server 端

from lacia.core.core import JsonRpc
from lacia.network.server.aioserver import AioServer, mount_app

namespace = {
    "ping": lambda x: f"pong {x}",
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

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="/ws")
    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 端

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)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

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="/ws")
    await rpc.run_client(client)

    Test = ProxyObj(rpc).Test

    assert await Test(1, 2).output("world") == "hello: world, a: 1, b: 2"

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, mount_app

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,
})

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

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, mount_app

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)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

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, Context
from lacia.core.proxy import ProxyObj
from lacia.network.server.aioserver import AioServer, mount_app

async def test_async_iter(n: int):
    name = Context.name.get()
    obj = ProxyObj(Context.rpc.get(), name)
    async for i in obj.test_async_iter(n):
        yield i

namespace = {
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(name = "server_test", namespace=namespace)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

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="client_test",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    proxy = ProxyObj(rpc).test_async_iter

    async for i in proxy(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, mount_app

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)

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

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())

Client to Client

Server 端

import asyncio
from lacia.core.core import JsonRpc, Context
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")

if __name__ == "__main__":

    from aiohttp import web
    app = web.Application()
    mount_app(app=app, server=server, rpc=rpc, path="/ws")
    web.run_app(app, host="localhost", port=8080)

Client 端 A

import asyncio
from lacia.core.core import JsonRpc
from lacia.core.proxy import ProxyObj
from lacia.network.client.aioclient import AioClient

async def test_async_iter(n: int):
    for i in range(n):
        await asyncio.sleep(1)
        yield i

namespace = {
    "ping": lambda x: f"pong {x}",
    "test_async_iter": test_async_iter,
}

rpc = JsonRpc(
    name="client_test_A",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.run_forever()

Client 端 B

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_B",
    namespace=namespace
)

async def main():

    client = AioClient(path="/ws")
    await rpc.run_client(client)

    c_obj = ProxyObj(rpc, "client_test_A")
    s_obj = ProxyObj(rpc)

    async for i in c_obj.test_async_iter(10):
        print(i)
    
    print(await s_obj.ping(c_obj.ping("hello")))
    print(await c_obj.ping(s_obj.ping("hello")))

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

Development Status
- 5 - Production/Stable
Operating System
- OS Independent
Programming Language
Typing
- Typed

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.6.tar.gz (16.1 kB view details)

Uploaded Apr 24, 2025 Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

The dropdown lists show the available interpreters, ABIs, and platforms. Enable javascript to be able to filter the list of wheel files.

lacia-0.2.6-py3-none-any.whl (21.2 kB view details)

Uploaded Apr 24, 2025 Python 3

File details

Details for the file lacia-0.2.6.tar.gz.

File metadata

Download URL: lacia-0.2.6.tar.gz
Upload date: Apr 24, 2025
Size: 16.1 kB
Tags: Source
Uploaded using Trusted Publishing? Yes
Uploaded via: pdm/2.24.1 CPython/3.13.3 Linux/6.11.0-1012-azure

File hashes

Hashes for lacia-0.2.6.tar.gz
Algorithm	Hash digest
SHA256	`f3ee4ca20a106ecafcaac2a2ce26e48f82ec7d904318a8ca074e100981b987f2`
MD5	`d07f9e6765208f1319b4c8ac143ff573`
BLAKE2b-256	`3dcf78d67e85387c800d75d1a75327faf3eb25eec5fd5407756ff64ffa9daefb`

See more details on using hashes here.

File details

Details for the file lacia-0.2.6-py3-none-any.whl.

File metadata

Download URL: lacia-0.2.6-py3-none-any.whl
Upload date: Apr 24, 2025
Size: 21.2 kB
Tags: Python 3
Uploaded using Trusted Publishing? Yes
Uploaded via: pdm/2.24.1 CPython/3.13.3 Linux/6.11.0-1012-azure

File hashes

Hashes for lacia-0.2.6-py3-none-any.whl
Algorithm	Hash digest
SHA256	`47c7af539cf743eb4b3ba86b66688776b11d7b00c2fc4e8482985af32095d1db`
MD5	`fa1773c01b5b0fd4b71056b242926c2e`
BLAKE2b-256	`ae742db81933751c46bf5bc2847f180b98d898dfb6ea9dca7b40e59a1e8433be`