cwkPippa 안의 Python — Walking Tour

# cwkPippa 의 Claude 어댑터 sketch (단순화)
from abc import ABC, abstractmethod
from typing import AsyncIterator
from dataclasses import dataclass

@dataclass
class StreamEvent:
    """어댑터가 caller 한테 yield 하는 거."""
    kind: str            # 'token' | 'tool_use' | 'thinking' | 'done'
    content: str | None = None

class Adapter(ABC):
    @abstractmethod
    async def stream(self, prompt: str) -> AsyncIterator[StreamEvent]:
        ...

class ClaudeAdapter(Adapter):
    async def stream(self, prompt: str) -> AsyncIterator[StreamEvent]:
        # 1. SDK 연결 (진짜 버전이 claude_agent_sdk 사용)
        # 2. prompt 보냄
        # 3. 청크 도착하면 yield
        # 4. yield 전 각 청크 JSONL 에 저장 (write-before-show)
        # 5. done 에 최종 마커 쓰고 return
        async for chunk in self._sdk_stream(prompt):
            yield StreamEvent(kind="token", content=chunk.text)
        yield StreamEvent(kind="done")

    async def _sdk_stream(self, prompt):
        # ... 실제 SDK 통합 ...
        yield  # 타입 체커용 placeholder

from fastapi import APIRouter
from pydantic import BaseModel

router = APIRouter()

class ChatRequest(BaseModel):
    conversation_id: str
    message: str
    brain: str = "claude"

class ChatChunk(BaseModel):
    kind: str
    content: str | None = None

@router.post("/api/chat")
async def chat(req: ChatRequest):
    """Server-Sent Event 로 chat 이벤트 stream."""
    # 1. 이전 세션의 깨진 turn 힐 (idempotent)
    # 2. 맞는 어댑터 (Claude / Codex / Gemini / Ollama)
    # 3. 이벤트 stream out, 각자 먼저 JSONL 에 저장
    # 4. done 에 JSONL 에서 최종 SQLite 행 derive
    # ...
    # 모양 — Pydantic in, async stream out, 중간에 JSONL ground truth
    pass

import json
from pathlib import Path
from datetime import datetime, timezone

class SessionLogger:
    """한 대화의 append-only JSONL writer."""

    def __init__(self, conversation_id: str, root: Path):
        self.path = root / f"{conversation_id}.jsonl"

    def append(self, event: dict) -> None:
        # Write before show — 사용자가 청크 보기 전 durability
        event["ts"] = datetime.now(timezone.utc).isoformat()
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(event, ensure_ascii=False) + "\n")

    def replay(self):
        # 복구 — JSONL 줄 단위로 읽어 SQLite 재빌드
        with self.path.open(encoding="utf-8") as f:
            for line in f:
                yield json.loads(line)

# 왜 JSONL? Append-only 쓰기, 줄 단위 streaming 읽기, 이벤트별 durable.
# Ground truth. SQLite + ChromaDB 가 derived mirror.

Foundations          → 변수, 타입, 문자열, 기본 I/O
Data                 → list, dict, tuple, set, 컴프리헨션, bytes
Flow                 → if/match, loop, 함수, closure, walrus
Iterators            → iter/next, generator, itertools, async iter
Decorators           → @, wraps, factory, 클래스 decorator, @property
OOP                  → class, dunder, dataclass, Protocol, ABC
OOP 고급             → MRO, mixin, dispatch, metaclass, descriptor
Errors               → try/except, EAFP, 커스텀 예외, contextmanager
Files & I/O          → open, pathlib, JSON, CSV, encoding, mmap
표준 라이브러리      → collections, functools, datetime, re, logging
Modules              → import, 패키지, venv, pyproject.toml
Typing               → 힌트, Literal, TypedDict, generic, Pydantic
Concurrency          → asyncio, threading, multiprocessing, GIL
Tooling              → pytest, mock, pdb, ruff, mypy, profiler
CLI                  → argparse, click, typer, rich
Pythonic             → EAFP, duck typing, 클래스 안 쓸 때, Zen

# cwkPippa 코드의 모든 줄이 이 list 의 뭔가 사용.
# 이제 다 읽을 수 있어.