railway-framework 0.8.2

Railway Oriented Programming framework for Python

Railway Framework for Python

型安全なパイプラインで、運用自動化をシンプルに。

# IDE補完が効く型安全なパイプライン
from railway import Contract, node, typed_pipeline

class UsersFetchResult(Contract):
    users: list[dict]
    total: int

class ReportResult(Contract):
    content: str

@node(output=UsersFetchResult)
def fetch_users() -> UsersFetchResult:
    return UsersFetchResult(users=[{"id": 1, "name": "Alice"}], total=1)

@node(inputs={"data": UsersFetchResult}, output=ReportResult)
def generate_report(data: UsersFetchResult) -> ReportResult:
    return ReportResult(content=f"{data.total} users found")
    #                            ^^^^
    #                            Ctrl+Space でフィールド補完！

result = typed_pipeline(fetch_users, generate_report)
print(result.content)  # IDE補完が効く！

特徴:

• IDE補完で開発効率アップ
• 型チェックでバグを早期発見
• テストはモック不要、引数を渡すだけ

---

Why Railway?

従来のパイプラインの問題

# ❌ 従来: 何が渡されるか分からない
def process(data):
    users = data["users"]  # KeyError? typo? IDE補完なし
    return {"processed": users[0]["name"]}  # ネストが深い...

result = pipeline(fetch, process, save)
# result["???"] 何が入ってる？

Railway の解決策

# ✅ Railway: 型契約で明確に
@node(inputs={"data": FetchResult}, output=ProcessResult)
def process(data: FetchResult) -> ProcessResult:
    users = data.users  # IDE補完 ✓ 型チェック ✓
    return ProcessResult(name=users[0].name)
    #                         ^^^^
    #                         Ctrl+Space で候補表示

result = typed_pipeline(fetch, process, save)
print(result.saved_count)  # 補完が効く！

観点	従来	Railway
データ構造	dict["key"]["nested"]	model.field
IDE補完	❌	✅
型チェック	❌	✅ (mypy対応)
テスト	モック必須	引数渡しのみ
リファクタ	grep検索	IDE一括変更 (F2)

---

Quick Start (5分)

1. インストール

# uvをインストール（未インストールの場合）
curl -LsSf https://astral.sh/uv/install.sh | sh

# railway コマンドをインストール
uv tool install railway-framework

2. プロジェクト作成

railway init my_automation
cd my_automation
uv sync
cp .env.example .env

3. 型契約（Contract）を定義

railway new contract UsersFetchResult

# src/contracts/users_fetch_result.py
from railway import Contract

class User(Contract):
    id: int
    name: str

class UsersFetchResult(Contract):
    users: list[User]
    total: int

4. 型付きノードを作成

railway new node fetch_users --output UsersFetchResult

# src/nodes/fetch_users.py
from railway import node
from contracts.users_fetch_result import UsersFetchResult, User

@node(output=UsersFetchResult)
def fetch_users() -> UsersFetchResult:
    # APIからユーザー取得
    return UsersFetchResult(
        users=[User(id=1, name="Alice")],
        total=1,
    )

5. テストを書く（TDD）

# tests/nodes/test_fetch_users.py
from nodes.fetch_users import fetch_users
from contracts.users_fetch_result import UsersFetchResult

def test_fetch_users():
    result = fetch_users()  # モック不要！

    assert isinstance(result, UsersFetchResult)
    assert result.total == len(result.users)

6. 実行

uv run railway run main

🎉 完成！ 型安全な自動化ツールができました。

---

アーキテクチャ

Contract（型契約）

ノード間で交換されるデータの「契約」を定義します。

from railway import Contract

class OrderResult(Contract):
    """注文処理の結果"""
    order_id: int
    status: str
    total: float

Contractの特徴:

• Pydantic BaseModel がベース（自動バリデーション）
• イミュータブル で安全（frozen=True）
• IDE補完 が効く

Node（処理単位）

@node(
    inputs={"order": OrderResult},  # 必要な入力を宣言
    output=ShippingResult,          # 出力の型を宣言
)
def create_shipping(order: OrderResult) -> ShippingResult:
    # 純粋関数として実装
    return ShippingResult(
        order_id=order.order_id,
        tracking_number=generate_tracking(),
    )

Pipeline（実行）

from railway import typed_pipeline

result = typed_pipeline(
    create_order,      # OrderResult を出力
    process_payment,   # PaymentResult を出力
    create_shipping,   # OrderResult を入力、ShippingResult を出力
)
# result は ShippingResult 型

依存関係はフレームワークが自動解決:

create_order ─────────────────┐
  output: OrderResult              │
                                   ├──> create_shipping
process_payment ──────────────┘       output: ShippingResult
  output: PaymentResult

---

CLI Commands

プロジェクト管理

railway init <name>              # プロジェクト作成
railway new entry <name>         # エントリポイント作成
railway docs                     # ドキュメント表示

Contract（型契約）

railway new contract <Name>          # Contract作成
railway new contract <Name> --entity # エンティティContract（id付き）
railway new contract <Name> --params # パラメータ用Contract
railway list contracts               # Contract一覧

Node（処理単位）

railway new node <name>                      # 基本node作成
railway new node <name> --output ResultType  # 出力型指定
railway new node <name> --input data:InputType --output ResultType
railway show node <name>                     # 依存関係表示

実行

railway run <entry>              # 実行
railway list                     # エントリポイント/ノード一覧

---

特徴

• ✨ 5分で開始: railway init でプロジェクト作成、すぐに実装開始
• 🛤️ 型安全パイプライン: Contract による型契約でIDE補完が効く
• 🔒 型チェック: mypyによる静的型チェック + ランタイム検証
• ⚡ 非同期対応: async/await 完全サポート
• 🎯 シンプルなAPI: デコレータベースで直感的
• 📝 自動生成: テンプレートから即座にコード生成
• 🧪 テスト容易: モック不要、引数を渡すだけ
• ⚙️ 環境別設定: development/production を簡単に切り替え
• 🔄 自動リトライ: 一時的なエラーに自動で対処
• 📊 構造化ロギング: loguru による美しいログ出力

---

コア概念

1. ノード (@node)

ノード = 再利用可能な処理単位

from railway import node
from loguru import logger

@node(retry=True)  # リトライ有効化
def fetch_data(url: str) -> dict:
    """データ取得ノード"""
    logger.info(f"Fetching from {url}")
    response = requests.get(url)
    return response.json()

型付きノード（推奨）:

@node(output=UsersFetchResult)
def fetch_users() -> UsersFetchResult:
    return UsersFetchResult(users=[...], total=10)

@node(inputs={"users": UsersFetchResult}, output=ReportResult)
def generate_report(users: UsersFetchResult) -> ReportResult:
    return ReportResult(content=f"{users.total} users")

2. エントリーポイント (@entry_point)

エントリーポイント = 実行の起点

from railway import entry_point, typed_pipeline

@entry_point
def main(date: str = None, dry_run: bool = False):
    """日次レポート生成"""
    result = typed_pipeline(
        fetch_data,
        process_data,
        generate_report,
    )
    return result

3. パイプライン

レガシーパイプライン（dict渡し）:

result = pipeline(
    step1(),
    step2,
    step3,
)

型付きパイプライン（推奨）:

result = typed_pipeline(
    fetch_users,      # UsersFetchResult を出力
    process_users,    # UsersFetchResult を入力
    generate_report,  # ReportResult を出力
)
# result は ReportResult 型

---

設定管理

統合設定ファイル: config/development.yaml

# アプリケーション設定
app:
  name: my_automation

# API設定
api:
  base_url: "https://api.example.com"
  timeout: 30

# ログ設定
logging:
  level: DEBUG
  handlers:
    - type: console
      level: DEBUG
    - type: file
      path: logs/app.log

# リトライ設定
retry:
  default:
    max_attempts: 3
    min_wait: 2
    max_wait: 10

コードから設定にアクセス

from src.settings import settings

url = settings.api.base_url
retry_config = settings.get_retry_settings("fetch_data")

---

テストの書き方

型付きノードはテストが簡単:

# tests/nodes/test_process_users.py
from contracts.users import UsersFetchResult, User
from contracts.report import ReportResult
from nodes.process_users import process_users

def test_process_users():
    # Arrange - 引数を渡すだけ（モック不要）
    users = UsersFetchResult(
        users=[User(id=1, name="Alice")],
        total=1,
    )

    # Act
    result = process_users(users)

    # Assert
    assert isinstance(result, ReportResult)
    assert "Alice" in result.content

# テスト実行
pytest -v
pytest --cov=src --cov-report=html

---

実例: 日次レポート生成

ステップ1: Contractを定義

railway new contract SalesData
railway new contract ReportResult

ステップ2: ノードを作成

railway new node fetch_sales --output SalesData
railway new node generate_report --input data:SalesData --output ReportResult

ステップ3: エントリーポイント

# src/daily_report.py
from railway import entry_point, typed_pipeline
from nodes.fetch_sales import fetch_sales
from nodes.generate_report import generate_report

@entry_point
def main(date: str = None):
    result = typed_pipeline(
        fetch_sales,
        generate_report,
    )
    print(result.content)
    return result

ステップ4: 実行

uv run railway run daily_report

---

非同期サポート

from railway import node
from railway.core.resolver import typed_async_pipeline

@node(output=UsersFetchResult)
async def fetch_users_async() -> UsersFetchResult:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            data = await response.json()
            return UsersFetchResult(users=data["users"], total=len(data["users"]))

@entry_point
async def main():
    result = await typed_async_pipeline(
        fetch_users_async,
        process_users,
    )
    return result

---

採用技術スタック

ライブラリ	用途
pydantic	Contract（データバリデーション）
tenacity	リトライ処理
typer	CLIインターフェース
loguru	構造化ロギング

---

ロードマップ

Phase 1 ✅ 完了

• ✅ @node, @entry_point デコレータ
• ✅ pipeline(), async_pipeline() 関数
• ✅ 設定管理、ロギング、リトライ
• ✅ CLIツール (init, new, list, run)

Phase 1.5 ✅ 完了（Output Model Pattern）

• ✅ Contract ベースクラス
• ✅ Params パラメータクラス
• ✅ typed_pipeline(), typed_async_pipeline()
• ✅ DependencyResolver 自動依存解決
• ✅ CLI拡張 (new contract, list contracts, show node)

Phase 2 📋 計画中

• 🔜 並列パイプライン実行
• 🔜 グラフベースワークフロー
• 🔜 WebUI
• 🔜 メトリクス収集

---

ライセンス

MIT License

---

Railway Framework で型安全な運用自動化を始めましょう！

railway init my_automation
cd my_automation
railway new contract UserResult
railway new node fetch_users --output UserResult
uv run railway run main