✍️ 필사 모드: Python 완전 가이드 — FastAPI·AsyncIO·Pydantic·uv·Polars·AI 엔지니어링 (Season 2 Ep 5, 2025)

python3.13t # free-threaded build

python --enable-experimental-jit

# PEP 695: Generic Syntax
def first[T](items: list[T]) -> T:
    return items[0]

class Stack[T]:
    def __init__(self) -> None:
        self.items: list[T] = []

# PEP 696: Type Parameter Defaults
def process[T = str](x: T) -> T: ...

# PEP 742: TypeIs
from typing import TypeIs
def is_str(x: object) -> TypeIs[str]:
    return isinstance(x, str)

# 설치
curl -LsSf https://astral.sh/uv/install.sh | sh

# 프로젝트 초기화
uv init my-app
cd my-app

# 의존성 추가 (pyproject.toml 자동 업데이트)
uv add fastapi uvicorn
uv add --dev pytest ruff mypy

# 실행
uv run python main.py
uv run pytest

# Python 버전 관리
uv python install 3.13
uv python pin 3.13

# Lock 파일 (uv.lock)
uv lock
uv sync

# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "httpx",
#     "rich",
# ]
# ///

import httpx
from rich import print

print(httpx.get("https://api.github.com").json())

uv run script.py
# 의존성 자동 다운로드 후 실행

from pydantic import BaseModel, EmailStr, Field
from typing import Annotated
from datetime import datetime

class User(BaseModel):
    id: int
    email: EmailStr
    name: Annotated[str, Field(min_length=1, max_length=100)]
    age: Annotated[int, Field(ge=0, le=150)]
    created_at: datetime

# 검증 + 변환
user = User.model_validate({
    "id": 1,
    "email": "alice@example.com",
    "name": "Alice",
    "age": 30,
    "created_at": "2025-01-01T00:00:00Z"
})

# 직렬화
json_str = user.model_dump_json()
dict_data = user.model_dump()

from pydantic import field_validator, model_validator

class Post(BaseModel):
    title: str
    content: str
    tags: list[str] = []

    @field_validator('title')
    @classmethod
    def title_not_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError('title cannot be empty')
        return v.strip()

    @model_validator(mode='after')
    def check_tags_limit(self) -> 'Post':
        if len(self.tags) > 10:
            raise ValueError('too many tags')
        return self

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    api_key: str
    debug: bool = False

    class Config:
        env_file = ".env"

settings = Settings()

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    email: str
    password: str

class UserResponse(BaseModel):
    id: int
    email: str

@app.post("/users", response_model=UserResponse)
async def create_user(user: UserCreate) -> UserResponse:
    # user는 이미 검증됨
    # 응답은 UserResponse로 자동 직렬화
    return UserResponse(id=1, email=user.email)

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        yield session

async def get_current_user(token: str = Header(...)) -> User:
    # 토큰 검증 로직
    return user

@app.get("/me")
async def me(user: User = Depends(get_current_user)) -> UserResponse:
    return user

import asyncio

async def fetch(url: str) -> str:
    # 시뮬레이션
    await asyncio.sleep(1)
    return f"data from {url}"

async def main() -> None:
    # 순차
    r1 = await fetch("a")
    r2 = await fetch("b")

    # 동시 (2배 빠름)
    r1, r2 = await asyncio.gather(fetch("a"), fetch("b"))

asyncio.run(main())

async def main():
    async with asyncio.TaskGroup() as tg:
        t1 = tg.create_task(fetch("a"))
        t2 = tg.create_task(fetch("b"))
        t3 = tg.create_task(fetch("c"))
    # 컨텍스트 나오면 모든 task 완료
    # 하나라도 예외면 나머지 취소 + ExceptionGroup
    print(t1.result(), t2.result(), t3.result())

async def fetch_with_timeout():
    async with asyncio.timeout(5.0):
        return await slow_operation()

import polars as pl

df = pl.read_csv("data.csv")

result = (
    df
    .filter(pl.col("age") > 30)
    .group_by("country")
    .agg([
        pl.col("salary").mean().alias("avg_salary"),
        pl.col("id").count().alias("count"),
    ])
    .sort("avg_salary", descending=True)
)

lazy_df = pl.scan_csv("huge.csv")  # 실제 읽지 않음

result = (
    lazy_df
    .filter(pl.col("year") == 2024)
    .group_by("category")
    .agg(pl.col("amount").sum())
    .collect()  # 이제 실행 (옵티마이저가 계획 수립)
)

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class State(TypedDict):
    messages: Annotated[list, operator.add]

def agent(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: State) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "tools"
    return END

workflow = StateGraph(State)
workflow.add_node("agent", agent)
workflow.add_node("tools", tool_node)
workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")

app = workflow.compile()

from pydantic_ai import Agent
from pydantic import BaseModel

class WeatherResponse(BaseModel):
    city: str
    temp_c: float
    condition: str

agent = Agent(
    'openai:gpt-4o',
    result_type=WeatherResponse,
    system_prompt='You are a weather assistant.',
)

result = await agent.run('Weather in Seoul?')
print(result.data)  # WeatherResponse 타입

# pyproject.toml
[project]
name = "my-app"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115",
    "pydantic>=2.9",
    "uvicorn>=0.30",
]

[dependency-groups]
dev = [
    "pytest>=8",
    "ruff>=0.7",
    "mypy>=1.13",
]

[tool.ruff]
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "S", "C4", "PL"]

[tool.mypy]
strict = true

# pytest + httpx for FastAPI
import pytest
from httpx import AsyncClient, ASGITransport
from myapp.main import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(
        transport=ASGITransport(app=app), base_url="http://test"
    ) as ac:
        r = await ac.post("/users", json={"email": "a@b.com", "password": "pw"})
        assert r.status_code == 200