pylego-blocks 0.1.1

Composable block-based Python component architecture framework

pylego

레고처럼 조립 가능한 파이썬 컴포넌트 아키텍처 프레임워크. 작은 표준 블록(Block)을 pydantic v2 계약(Stud)으로 스냅해 Assembly로 조립한다.

설치

GitHub Releases (Private)

GITHUB_TOKEN은 repo 권한이 필요합니다.

wheel 직접 설치:

pip install "https://<GITHUB_TOKEN>@github.com/hoksi/pylego/releases/download/v0.1.0/pylego-0.1.0-py3-none-any.whl"

git 태그로 설치:

pip install "git+https://<GITHUB_TOKEN>@github.com/hoksi/pylego.git@v0.1.0"

uv 사용 시:

uv add "pylego @ git+https://<GITHUB_TOKEN>@github.com/hoksi/pylego.git@v0.1.0"

개발용 (소스 직접)

git clone https://github.com/hoksi/pylego.git
cd pylego
uv sync

---

빠른 시작

1. Block 정의

from pylego.core.block import Block
from pylego.core.stud import Stud


class TextInput(Stud):
    text: str


class UpperOutput(Stud):
    text: str


class LengthOutput(Stud):
    length: int


class UpperBlock(Block[TextInput, UpperOutput]):
    input_model = TextInput
    output_model = UpperOutput

    async def run(self, inp: TextInput) -> UpperOutput:
        return UpperOutput(text=inp.text.upper())


class LengthBlock(Block[UpperOutput, LengthOutput]):
    input_model = UpperOutput
    output_model = LengthOutput

    async def run(self, inp: UpperOutput) -> LengthOutput:
        return LengthOutput(length=len(inp.text))

2. Assembly 조립 및 실행

from pylego.core.assembly import Assembly

# 조립 시점에 Stud 계약 불일치를 즉시 검출
pipeline = Assembly([UpperBlock(), LengthBlock()])

# 비동기
import asyncio
result = asyncio.run(pipeline.run(TextInput(text="hello")))

# 동기 편의 API
result = pipeline.run_sync(TextInput(text="hello"))
print(result.length)  # 5

3. YAML 선언형 조립

# pipeline.yaml
name: text-pipeline
blocks:
  - class: myapp.blocks.UpperBlock
  - class: myapp.blocks.LengthBlock

from pylego.declarative import load_assembly

pipeline = load_assembly("pipeline.yaml")
result = pipeline.run_sync(TextInput(text="hello"))

4. FanOut — 동시 분기 실행

from pylego.core.fanout import FanOut, Gathered

fan = FanOut([UpperBlock(), LengthBlock()])
# 두 브랜치를 asyncio.gather 로 동시 실행 → Gathered(results=(…, …))
result: Gathered = asyncio.run(fan.run(TextInput(text="hello")))

5. BlockRegistry

from pylego.registry import register, default_registry

@register
class UpperBlock(Block[TextInput, UpperOutput]):
    ...

cls = default_registry.get("UpperBlock")

---

핵심 원칙

1. 블록은 다른 블록 내부를 모른다 — Stud(계약)만 안다.
2. 블록은 단독 실행·단독 테스트가 가능해야 한다.
3. Assembly는 블록과 구분 불가능해야 한다 (재귀적 합성).
4. 런타임 교체(sync↔async)가 블록 수정 없이 가능해야 한다.

---

개발 환경

# 의존성 설치
uv sync

# 테스트
uv run pytest

# 타입 체크
uv run mypy src/

# 빌드
uv build

문서

• 로드맵
• CHANGELOG