cordis-py
Cordis(A Meta-Framework of Spatiotemporal Composability)的 Python 移植。
核心概念与 TS 版 1:1 对应:Context(依赖注入容器)、Fiber(作用域化副作用管理 + 依赖重载)、Registry(插件注册)、Events(五种分发模式)、Service(服务基类)。
状态:核心移植完成(80 测试全绿)。上游 MIT 许可证。 设计与偏离记录:
../docs/design_notes/20260816-python-port-architecture.md与COMPAT.md。
安装
pip install cordis-py # 核心(零运行时依赖)
pip install "cordis-py[pydantic]" # 可选:Pydantic 配置校验
要求 Python ≥ 3.10。
mkdir -p thirdparty
cd thirdparty
git clone https://github.com/cordiverse/cordis.git
快速上手
import asyncio
from cordis import Context
async def main(ctx: Context):
# 事件
ctx.on('greet', lambda name: print(f'hello {name}'))
ctx.emit('greet', 'world')
# 插件(函数 / 类 / 带 apply 的对象三形态)
await ctx.plugin(logger_plugin, {'level': 'info'})
def logger_plugin(ctx, config):
def disposer():
print('plugin disposed')
print(f'logger plugin with config: {config}')
return disposer
Context().run(main(Context()))
核心概念
插件与依赖注入
from cordis import Context, Service, inject
class Database(Service):
# 服务名 'database',构造即注册
def __init__(self, ctx, config=None):
super().__init__(ctx, 'database')
self.url = config['url']
class App(Service):
inject = ['database'] # 依赖声明:database 可用后自动激活
def __init__(self, ctx):
super().__init__(ctx, 'app')
@inject('database') # 方法级注入:依赖出现后执行一次
def on_database_ready(self):
print('database ready')
return lambda: print('database down')
async def main(ctx: Context):
await ctx.plugin(Database, {'url': 'sqlite://:memory:'})
await ctx.plugin(App)
# database 提供 → App 自动激活 → on_database_ready 执行
ctx = Context()
ctx.run(main(ctx))
服务作用域(isolate)与事件隔离
child = ctx.isolate('database') # 隔离的服务空间
service_ctx = ctx.isolate('db')
service_ctx.on('db/connect', on_connect) # 只在同隔离内触发
配置校验(可选 Pydantic)
from pydantic import BaseModel
class DatabaseConfig(BaseModel):
url: str
pool_size: int = 5
class Database(Service):
Config = DatabaseConfig
# 插件 config 经 Pydantic 校验,失败抛 ValidationError
五种事件分发模式
| 模式 | 语义 |
|---|---|
ctx.emit(name, ...) |
同步广播 |
await ctx.parallel(name, ...) |
并发执行,失败聚合 ExceptionGroup |
await ctx.serial(name, ...) |
串行,首个非空结果短路 |
ctx.bail(name, ...) |
同步短路 |
ctx.waterfall(name, ...) |
管道,next 串接 |
与上游的差异
见 COMPAT.md。核心差异:shadow/caller 机制裁剪(Python 无隐式 this)、
Context.is_(is 保留字)、root 级属性访问返回 None(对齐 undefined)、
Pydantic 配置校验(可选 extra)。
开发
cd python
.venv/bin/python -m pytest tests/ # 80 测试
.venv/bin/ruff check src/ tests/ # lint
Metadata
Release files for pi-cordis-py 0.1.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| pi_cordis_py-0.1.0.tar.gz | 145.2 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| pi_cordis_py-0.1.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 174.9 kB
Release files / pi_cordis_py-0.1.0.tar.gz
| Download URL | pi_cordis_py-0.1.0.tar.gz |
|---|---|
| Size | 145.2 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
d929dc3b7af14c8dd56bcb590593cdae137fa86ab07c13039c5479eb6673ae46
|
|
BLAKE2b-256 checksum How to use checksums |
43ff76ef5554759adf9c1a1ee01221f5c0256f7ade62675808449ec3b0791b93
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.11.15
|
Release files / pi_cordis_py-0.1.0-py3-none-any.whl
| Download URL | pi_cordis_py-0.1.0-py3-none-any.whl |
|---|---|
| Size | 29.6 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
a420cd9fd3a35b7f2f6aa98b6792527bc7f49b644c0f6c8efc960eb104e13157
|
|
BLAKE2b-256 checksum How to use checksums |
37600b7fe95a8731564465c2174ecdeaa32bc77619ede213ee014eb3594a4134
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.11.15
|