Split View: Python 베스트 프랙티스 & 고급 패턴 가이드 — 프로처럼 파이썬 쓰기

my-project/
  src/
    my_package/
      __init__.py
      core.py
      utils.py
  tests/
    test_core.py
    test_utils.py
  pyproject.toml
  README.md

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[project]
name = "my-package"
version = "0.1.0"
description = "My awesome package"
requires-python = ">=3.11"
dependencies = [
    "httpx>=0.27",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "mypy>=1.8",
    "ruff>=0.3",
]

[tool.ruff]
line-length = 88
target-version = "py311"

[tool.mypy]
strict = true
python_version = "3.11"

python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"

uv venv
uv pip install -e ".[dev]"
# 또는 uv sync로 lockfile 기반 설치
uv sync

# Python 3.10+에서는 typing 없이 가능
def greet(name: str) -> str:
    return f"Hello, {name}!"

def process_items(items: list[str]) -> dict[str, int]:
    return {item: len(item) for item in items}

# Union 타입도 | 연산자로 표현
def parse_value(value: str | int | None) -> str:
    if value is None:
        return "empty"
    return str(value)

from typing import TypeVar, Generic

T = TypeVar("T")

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

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items.pop()

    def peek(self) -> T:
        if not self._items:
            raise IndexError("Stack is empty")
        return self._items[-1]

# 사용
int_stack: Stack[int] = Stack()
int_stack.push(42)
str_stack: Stack[str] = Stack()
str_stack.push("hello")

from typing import Protocol, runtime_checkable

@runtime_checkable
class Drawable(Protocol):
    def draw(self) -> str: ...

class Circle:
    def draw(self) -> str:
        return "Drawing circle"

class Square:
    def draw(self) -> str:
        return "Drawing square"

def render(shape: Drawable) -> None:
    print(shape.draw())

# Circle은 Drawable을 상속하지 않지만 draw()가 있으므로 호환됨
render(Circle())  # OK
render(Square())  # OK

[tool.mypy]
strict = true
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true

[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false

# 전체 프로젝트 검사
mypy src/
# 특정 파일 검사
mypy src/my_package/core.py

import functools
import time
from typing import Callable, ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")

def timer(func: Callable[P, R]) -> Callable[P, R]:
    """함수 실행 시간을 측정하는 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function(n: int) -> int:
    """n까지의 합"""
    return sum(range(n))

slow_function(1_000_000)
# 출력: slow_function took 0.0312s

def retry(max_attempts: int = 3, delay: float = 1.0):
    """실패 시 재시도하는 데코레이터"""
    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            last_exception: Exception | None = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    print(f"Attempt {attempt} failed: {e}")
                    if attempt < max_attempts:
                        time.sleep(delay)
            raise last_exception  # type: ignore
        return wrapper
    return decorator

@retry(max_attempts=5, delay=2.0)
def fetch_data(url: str) -> dict:
    # 네트워크 요청 로직
    ...

class CacheResult:
    """결과를 캐싱하는 클래스 데코레이터"""
    def __init__(self, ttl_seconds: int = 300):
        self.ttl = ttl_seconds
        self.cache: dict[str, tuple[float, object]] = {}

    def __call__(self, func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            key = str(args) + str(kwargs)
            now = time.time()
            if key in self.cache:
                cached_time, cached_result = self.cache[key]
                if now - cached_time < self.ttl:
                    return cached_result  # type: ignore
            result = func(*args, **kwargs)
            self.cache[key] = (now, result)
            return result
        return wrapper

@CacheResult(ttl_seconds=60)
def expensive_computation(x: int) -> int:
    time.sleep(2)  # 비싼 연산 시뮬레이션
    return x ** 2

class DatabaseConnection:
    def __init__(self, connection_string: str):
        self.connection_string = connection_string
        self.connection = None

    def __enter__(self):
        print(f"Connecting to {self.connection_string}")
        self.connection = self._connect()
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.connection:
            self.connection.close()
            print("Connection closed")
        # False를 반환하면 예외가 전파됨
        return False

    def _connect(self):
        # 실제 연결 로직
        ...

with DatabaseConnection("postgresql://localhost/mydb") as conn:
    conn.execute("SELECT 1")
# 블록을 나가면 자동으로 연결이 닫힘

from contextlib import contextmanager
import os

@contextmanager
def temporary_directory(path: str):
    """임시 디렉터리를 생성하고 사용 후 삭제"""
    os.makedirs(path, exist_ok=True)
    try:
        yield path
    finally:
        import shutil
        shutil.rmtree(path)

with temporary_directory("/tmp/work") as tmpdir:
    # tmpdir을 사용한 작업
    print(f"Working in {tmpdir}")
# 블록이 끝나면 디렉터리 자동 삭제

from contextlib import asynccontextmanager
import aiohttp

@asynccontextmanager
async def http_session():
    """aiohttp 세션 관리"""
    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()

async def fetch_url(url: str) -> str:
    async with http_session() as session:
        async with session.get(url) as response:
            return await response.text()

# 리스트: 모든 데이터를 메모리에 로드
numbers_list = [x ** 2 for x in range(10_000_000)]  # ~80MB

# 제너레이터: 하나씩 생성, 거의 메모리 사용 없음
numbers_gen = (x ** 2 for x in range(10_000_000))  # ~120B

def read_large_file(filepath: str, chunk_size: int = 8192):
    """대용량 파일을 청크 단위로 읽기"""
    with open(filepath, "r") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            yield chunk

# 수 GB 파일도 메모리 걱정 없이 처리
for chunk in read_large_file("huge_log.txt"):
    process(chunk)

def flatten(nested: list) -> list:
    """중첩 리스트를 평탄화"""
    for item in nested:
        if isinstance(item, list):
            yield from flatten(item)
        else:
            yield item

data = [1, [2, 3], [4, [5, 6]], 7]
print(list(flatten(data)))
# 출력: [1, 2, 3, 4, 5, 6, 7]

def chain(*iterables):
    """여러 이터러블을 하나로 연결"""
    for iterable in iterables:
        yield from iterable

result = list(chain([1, 2], [3, 4], [5, 6]))
# 출력: [1, 2, 3, 4, 5, 6]

import csv
from typing import Iterator

def read_csv_rows(filename: str) -> Iterator[dict]:
    """CSV 파일을 행 단위로 읽기"""
    with open(filename) as f:
        reader = csv.DictReader(f)
        yield from reader

def filter_active(rows: Iterator[dict]) -> Iterator[dict]:
    """활성 사용자만 필터링"""
    for row in rows:
        if row["status"] == "active":
            yield row

def extract_emails(rows: Iterator[dict]) -> Iterator[str]:
    """이메일 필드 추출"""
    for row in rows:
        yield row["email"]

# 파이프라인: 파일 -> 필터 -> 변환
pipeline = extract_emails(filter_active(read_csv_rows("users.csv")))
for email in pipeline:
    send_newsletter(email)

import asyncio

async def fetch_data(url: str, delay: float) -> str:
    print(f"Fetching {url}...")
    await asyncio.sleep(delay)  # 네트워크 요청 시뮬레이션
    return f"Data from {url}"

async def main():
    # 순차 실행: 3초
    result1 = await fetch_data("api/users", 1.0)
    result2 = await fetch_data("api/posts", 1.0)
    result3 = await fetch_data("api/comments", 1.0)

    # 병렬 실행: 1초
    results = await asyncio.gather(
        fetch_data("api/users", 1.0),
        fetch_data("api/posts", 1.0),
        fetch_data("api/comments", 1.0),
    )
    print(results)

asyncio.run(main())

import aiohttp
import asyncio

async def fetch_url(session: aiohttp.ClientSession, url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

async def fetch_all_users(user_ids: list[int]) -> list[dict]:
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_url(session, f"https://api.example.com/users/{uid}")
            for uid in user_ids
        ]
        return await asyncio.gather(*tasks)

# 100명의 사용자 정보를 병렬로 가져오기
users = asyncio.run(fetch_all_users(list(range(1, 101))))

async def rate_limited_fetch(
    urls: list[str],
    max_concurrent: int = 10,
) -> list[str]:
    """동시 요청 수를 제한하는 함수"""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def fetch_with_limit(session: aiohttp.ClientSession, url: str) -> str:
        async with semaphore:
            async with session.get(url) as resp:
                return await resp.text()

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, url) for url in urls]
        return await asyncio.gather(*tasks)

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(fetch_data("api/users", 1.0))
        task2 = tg.create_task(fetch_data("api/posts", 1.0))
        task3 = tg.create_task(fetch_data("api/comments", 1.0))

    # 모든 태스크가 완료되면 여기에 도달
    print(task1.result(), task2.result(), task3.result())

from dataclasses import dataclass, field

@dataclass
class User:
    name: str
    email: str
    age: int
    tags: list[str] = field(default_factory=list)
    is_active: bool = True

    def display_name(self) -> str:
        return f"{self.name} ({self.email})"

user = User(name="Kim", email="kim@example.com", age=30)
print(user)
# User(name='Kim', email='kim@example.com', age=30, tags=[], is_active=True)

@dataclass(frozen=True)
class Point:
    x: float
    y: float

p = Point(1.0, 2.0)
# p.x = 3.0  # FrozenInstanceError 발생!

from typing import NamedTuple

class Coordinate(NamedTuple):
    latitude: float
    longitude: float
    altitude: float = 0.0

coord = Coordinate(37.5665, 126.9780)
lat, lng, alt = coord  # 언패킹 가능
print(coord.latitude)  # 속성 접근

from pydantic import BaseModel, EmailStr, Field, field_validator

class UserCreate(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    email: EmailStr
    age: int = Field(ge=0, le=150)
    tags: list[str] = []

    @field_validator("name")
    @classmethod
    def name_must_be_capitalized(cls, v: str) -> str:
        if not v[0].isupper():
            raise ValueError("Name must start with uppercase")
        return v

# 올바른 입력
user = UserCreate(name="Kim", email="kim@example.com", age=30)
print(user.model_dump_json())

# 잘못된 입력 -> ValidationError
try:
    bad_user = UserCreate(name="", email="not-an-email", age=-5)
except Exception as e:
    print(e)

class DatabasePool:
    _instance: "DatabasePool | None" = None

    def __new__(cls) -> "DatabasePool":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._initialize()
        return cls._instance

    def _initialize(self) -> None:
        self.connections: list = []
        print("Pool initialized")

# 항상 같은 인스턴스를 반환
pool1 = DatabasePool()
pool2 = DatabasePool()
assert pool1 is pool2

# db.py
_pool = None

def get_pool():
    global _pool
    if _pool is None:
        _pool = create_pool()
    return _pool

from abc import ABC, abstractmethod

class Notification(ABC):
    @abstractmethod
    def send(self, message: str) -> None: ...

class EmailNotification(Notification):
    def send(self, message: str) -> None:
        print(f"Email: {message}")

class SlackNotification(Notification):
    def send(self, message: str) -> None:
        print(f"Slack: {message}")

class SMSNotification(Notification):
    def send(self, message: str) -> None:
        print(f"SMS: {message}")

def create_notification(channel: str) -> Notification:
    factories: dict[str, type[Notification]] = {
        "email": EmailNotification,
        "slack": SlackNotification,
        "sms": SMSNotification,
    }
    if channel not in factories:
        raise ValueError(f"Unknown channel: {channel}")
    return factories[channel]()

notif = create_notification("slack")
notif.send("Hello!")

from typing import Callable

class EventEmitter:
    def __init__(self) -> None:
        self._listeners: dict[str, list[Callable]] = {}

    def on(self, event: str, callback: Callable) -> None:
        self._listeners.setdefault(event, []).append(callback)

    def emit(self, event: str, *args, **kwargs) -> None:
        for callback in self._listeners.get(event, []):
            callback(*args, **kwargs)

# 사용
emitter = EventEmitter()
emitter.on("user_created", lambda user: print(f"Welcome, {user}!"))
emitter.on("user_created", lambda user: send_email(user))
emitter.emit("user_created", "Kim")

from typing import Protocol

class SortStrategy(Protocol):
    def sort(self, data: list[int]) -> list[int]: ...

class BubbleSort:
    def sort(self, data: list[int]) -> list[int]:
        arr = data.copy()
        n = len(arr)
        for i in range(n):
            for j in range(0, n - i - 1):
                if arr[j] > arr[j + 1]:
                    arr[j], arr[j + 1] = arr[j + 1], arr[j]
        return arr

class QuickSort:
    def sort(self, data: list[int]) -> list[int]:
        if len(data) <= 1:
            return data
        pivot = data[len(data) // 2]
        left = [x for x in data if x < pivot]
        middle = [x for x in data if x == pivot]
        right = [x for x in data if x > pivot]
        return self.sort(left) + middle + self.sort(right)

class Sorter:
    def __init__(self, strategy: SortStrategy) -> None:
        self._strategy = strategy

    def sort(self, data: list[int]) -> list[int]:
        return self._strategy.sort(data)

# 전략을 런타임에 교체
sorter = Sorter(QuickSort())
print(sorter.sort([3, 1, 4, 1, 5, 9]))

import cProfile
import pstats

def expensive_function():
    total = 0
    for i in range(1_000_000):
        total += i ** 2
    return total

# 프로파일링 실행
profiler = cProfile.Profile()
profiler.enable()
expensive_function()
profiler.disable()

# 결과 출력
stats = pstats.Stats(profiler)
stats.sort_stats("cumulative")
stats.print_stats(10)

pip install line_profiler

@profile  # line_profiler 데코레이터
def process_data(data: list[int]) -> list[int]:
    result = []                    # 거의 0
    for item in data:              # 루프 오버헤드
        if item % 2 == 0:         # 조건 검사
            result.append(item * 2)  # 실제 작업
    return result

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

# 캐시 없이: O(2^n) -> 매우 느림
# 캐시 있으면: O(n) -> 즉시 완료
print(fibonacci(100))

# 캐시 통계 확인
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)

import timeit

# for 루프 - 느림
def squares_loop(n: int) -> list[int]:
    result = []
    for i in range(n):
        result.append(i ** 2)
    return result

# 리스트 컴프리헨션 - 빠름 (C 레벨 최적화)
def squares_comp(n: int) -> list[int]:
    return [i ** 2 for i in range(n)]

# 벤치마크
n = 100_000
t_loop = timeit.timeit(lambda: squares_loop(n), number=100)
t_comp = timeit.timeit(lambda: squares_comp(n), number=100)
print(f"Loop: {t_loop:.3f}s, Comprehension: {t_comp:.3f}s")
# 컴프리헨션이 보통 20~30% 빠릅니다

# 1. 문자열 결합은 join() 사용
words = ["hello", "world", "python"]
result = " ".join(words)  # "+" 연산자보다 훨씬 빠름

# 2. 멤버십 검사는 set 사용
valid_ids = set(range(10000))
if 42 in valid_ids:  # O(1) - 리스트의 O(n)보다 빠름
    pass

# 3. collections 활용
from collections import Counter, defaultdict

# 빈도 계산
word_count = Counter(["apple", "banana", "apple", "cherry", "apple"])
print(word_count.most_common(2))  # [('apple', 3), ('banana', 1)]

# 기본값 딕셔너리
grouped = defaultdict(list)
for item in [("A", 1), ("B", 2), ("A", 3)]:
    grouped[item[0]].append(item[1])

# tests/test_calculator.py
def add(a: int, b: int) -> int:
    return a + b

def test_add_positive():
    assert add(2, 3) == 5

def test_add_negative():
    assert add(-1, -1) == -2

def test_add_zero():
    assert add(0, 0) == 0

# 테스트 실행
pytest tests/ -v
# 커버리지 포함
pytest --cov=src --cov-report=term-missing

import pytest

@pytest.fixture
def sample_users() -> list[dict]:
    return [
        {"name": "Kim", "age": 30},
        {"name": "Lee", "age": 25},
        {"name": "Park", "age": 35},
    ]

@pytest.fixture
def db_connection():
    conn = create_connection()
    yield conn  # 테스트 실행
    conn.close()  # 정리 코드

def test_user_count(sample_users):
    assert len(sample_users) == 3

def test_oldest_user(sample_users):
    oldest = max(sample_users, key=lambda u: u["age"])
    assert oldest["name"] == "Park"

@pytest.mark.parametrize("input_val, expected", [
    ("hello", 5),
    ("", 0),
    ("python", 6),
    ("   ", 3),
])
def test_string_length(input_val: str, expected: int):
    assert len(input_val) == expected

@pytest.mark.parametrize("a, b, expected", [
    (2, 3, 5),
    (-1, 1, 0),
    (0, 0, 0),
    (100, 200, 300),
])
def test_add(a: int, b: int, expected: int):
    assert add(a, b) == expected

from unittest.mock import Mock, patch, AsyncMock

class UserService:
    def __init__(self, api_client):
        self.api = api_client

    def get_user(self, user_id: int) -> dict:
        response = self.api.get(f"/users/{user_id}")
        return response.json()

def test_get_user():
    # Mock API 클라이언트 생성
    mock_api = Mock()
    mock_api.get.return_value.json.return_value = {
        "id": 1,
        "name": "Kim",
    }

    service = UserService(mock_api)
    user = service.get_user(1)

    assert user["name"] == "Kim"
    mock_api.get.assert_called_once_with("/users/1")

# patch를 사용한 모킹
@patch("my_module.requests.get")
def test_fetch_data(mock_get):
    mock_get.return_value.status_code = 200
    mock_get.return_value.json.return_value = {"data": "test"}

    result = fetch_data("https://api.example.com")
    assert result["data"] == "test"

import pytest

@pytest.mark.asyncio
async def test_async_fetch():
    result = await fetch_data("https://api.example.com/users/1")
    assert "name" in result

@pytest.mark.asyncio
async def test_concurrent_requests():
    results = await asyncio.gather(
        fetch_data("url1"),
        fetch_data("url2"),
    )
    assert len(results) == 2