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
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file pi_cordis_py-0.1.0.tar.gz.
File metadata
- Download URL: pi_cordis_py-0.1.0.tar.gz
- Upload date:
- Size: 145.2 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
d929dc3b7af14c8dd56bcb590593cdae137fa86ab07c13039c5479eb6673ae46
|
|
| MD5 |
e93187a1e096eacfe228c45a50626e30
|
|
| BLAKE2b-256 |
43ff76ef5554759adf9c1a1ee01221f5c0256f7ade62675808449ec3b0791b93
|
File details
Details for the file pi_cordis_py-0.1.0-py3-none-any.whl.
File metadata
- Download URL: pi_cordis_py-0.1.0-py3-none-any.whl
- Upload date:
- Size: 29.6 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/7.0.0 CPython/3.11.15
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
a420cd9fd3a35b7f2f6aa98b6792527bc7f49b644c0f6c8efc960eb104e13157
|
|
| MD5 |
de104247ca47f7395b9d4e7744768f5e
|
|
| BLAKE2b-256 |
37600b7fe95a8731564465c2174ecdeaa32bc77619ede213ee014eb3594a4134
|