# cordis-py

**Repository Path**: pi-lab/cordis-py

## Basic Information

- **Project Name**: cordis-py
- **Description**: Meta-Framework of Spatiotemporal Composability for Python


- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-08-16
- **Last Updated**: 2026-08-18

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# cordis-py

Cordis（[A Meta-Framework of Spatiotemporal Composability](https://github.com/cordiverse/cordis)）的 Python 移植。

核心概念与 TS 版 1:1 对应：**Context**（依赖注入容器）、**Fiber**（作用域化副作用管理 + 依赖重载）、**Registry**（插件注册）、**Events**（五种分发模式）、**Service**（服务基类）。

> 状态：核心移植完成（80 测试全绿）。上游 MIT 许可证。
> 设计与偏离记录：`../docs/design_notes/20260816-python-port-architecture.md` 与 `COMPAT.md`。

## 安装

```bash
pip install cordis-py                      # 核心（零运行时依赖）
pip install "cordis-py[pydantic]"         # 可选：Pydantic 配置校验
```

要求 Python ≥ 3.10。

```bash
mkdir -p thirdparty
cd thirdparty
git clone https://github.com/cordiverse/cordis.git
```

## 快速上手

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

## 核心概念

### 插件与依赖注入

```python
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）与事件隔离

```python
child = ctx.isolate('database')   # 隔离的服务空间
service_ctx = ctx.isolate('db')
service_ctx.on('db/connect', on_connect)   # 只在同隔离内触发
```

### 配置校验（可选 Pydantic）

```python
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`](docs/COMPAT.md)。核心差异：shadow/caller 机制裁剪（Python 无隐式 this）、
`Context.is_`（`is` 保留字）、root 级属性访问返回 `None`（对齐 undefined）、
Pydantic 配置校验（可选 extra）。

## 开发

```bash
cd python
.venv/bin/python -m pytest tests/   # 80 测试
.venv/bin/ruff check src/ tests/    # lint
```