Advanced Exercises: Chapter 9

Chapter 9: Functions - 심화 연습문제

고급 함수 패턴, 클로저, 고급 데코레이터, 함수형 프로그래밍

난이도: ★★★ (고급) ~ ★★★★ (도전)

문제 1^★★★: Mutable Default Arguments의 함정

Python에서 mutable 객체를 기본 인자로 사용할 때 발생하는 문제를 이해하고 해결책을 구현하세요.

Part 1: 문제 재현

다음 코드의 문제점을 찾고 왜 이런 현상이 발생하는지 설명하세요:

def add_item(item, item_list=[]):
    item_list.append(item)
    return item_list

# 테스트
print(add_item("apple"))      # ['apple']
print(add_item("banana"))     # ['apple', 'banana'] - 예상과 다름!
print(add_item("cherry"))     # ['apple', 'banana', 'cherry']

Part 2: 해결책 구현

3가지 다른 방법으로 이 문제를 해결하세요:

1. 방법 1: None을 기본값으로 사용
2. 방법 2: Sentinel 값 사용
3. 방법 3: 함수 속성을 활용한 캐시 초기화

Part 3: 실전 응용

다음 시나리오에서 올바른 기본 인자 처리를 구현하세요:

def create_user_profile(name, age, skills=[], preferences={}):
    """사용자 프로필 생성 - 버그가 있는 버전"""
    profile = {
        'name': name,
        'age': age,
        'skills': skills,
        'preferences': preferences
    }
    return profile

# 이 함수를 고치고, 테스트 케이스를 작성하세요

도전 과제: 기본 인자가 호출될 때마다 “새로 생성”되는 것처럼 동작하는 데코레이터를 만드세요.

# 여기에 코드를 작성하세요

문제 2^★★★: 파라미터를 받는 Decorator 구현

파라미터를 받아 동작을 제어하는 고급 데코레이터를 구현하세요.

구현할 데코레이터:

@retry(max_attempts=3, delay=1.0, exceptions=(ValueError, TypeError))
def unreliable_api_call(data):
    """가끔 실패하는 API 호출 시뮬레이션"""
    import random
    if random.random() < 0.7:  # 70% 확률로 실패
        raise ValueError("API call failed")
    return f"Success: {data}"

@log_calls(log_file="function_calls.log", include_result=True)
def calculate(x, y):
    return x + y

@rate_limit(calls_per_second=2)
def api_request(endpoint):
    """초당 2번만 호출 가능"""
    return f"Requesting {endpoint}"

@validate_types(x=int, y=int, return_type=int)
def add(x, y):
    return x + y

요구사항:

1. @retry: 실패 시 자동으로 재시도
   • max_attempts: 최대 시도 횟수
   • delay: 재시도 간 대기 시간
   • exceptions: 재시도할 예외 타입들
   • 매 시도마다 메시지 출력
2. @log_calls: 함수 호출 로깅
   • 함수 이름, 인자, 실행 시간 기록
   • 선택적으로 결과값도 기록
   • 파일에 저장
3. @rate_limit: 호출 빈도 제한
   • 지정된 횟수 이상 호출 시 대기
   • 마지막 호출 시간 추적
4. @validate_types: 타입 검증
   • 입력 파라미터 타입 체크
   • 반환값 타입 체크
   • 타입 불일치 시 TypeError 발생

힌트:

def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        def wrapper(*args, **kwargs):
            # 여기에 재시도 로직 구현
            pass
        return wrapper
    return decorator

# 여기에 코드를 작성하세요

문제 3^★★★: 클로저와 nonlocal을 활용한 상태 관리

클로저를 사용하여 상태를 캡슐화하는 여러 패턴을 구현하세요.

Part 1: 카운터 팩토리

def make_counter(start=0, step=1):
    """
    카운터 함수를 생성하는 팩토리
    
    반환되는 함수는 다음 메서드를 가져야 함:
    - __call__(): 현재 값을 반환하고 증가
    - reset(): 카운터를 start로 초기화
    - get(): 현재 값만 반환 (증가 없이)
    - set(value): 카운터를 특정 값으로 설정
    """
    # 구현

# 테스트
counter1 = make_counter(0, 1)
print(counter1())  # 0
print(counter1())  # 1
print(counter1())  # 2
counter1.reset()
print(counter1())  # 0

counter2 = make_counter(10, 5)
print(counter2())  # 10
print(counter2())  # 15

Part 2: 통계 추적기

def make_statistics_tracker():
    """
    숫자를 추적하고 통계를 제공하는 함수 생성
    
    반환되는 객체는:
    - add(value): 값 추가
    - mean(): 평균 반환
    - std(): 표준편차 반환
    - min(), max(): 최솟값, 최댓값
    - count(): 총 개수
    - clear(): 모든 데이터 초기화
    """
    # 구현

# 테스트
stats = make_statistics_tracker()
stats.add(10)
stats.add(20)
stats.add(30)
print(stats.mean())   # 20.0
print(stats.count())  # 3

Part 3: 함수 체이닝

def make_chainable_calculator(initial=0):
    """
    메서드 체이닝이 가능한 계산기
    
    예시:
    calc = make_chainable_calculator(10)
    result = calc.add(5).multiply(2).subtract(3).divide(3).value()
    # (10 + 5) * 2 - 3) / 3 = 9.0
    """
    # 구현

도전 과제: 각 계산 단계를 되돌릴 수 있는 undo() 메서드를 추가하세요.

# 여기에 코드를 작성하세요

문제 4^★★★★: 고급 Generator - Coroutine 패턴

Generator의 send(), throw(), close() 메서드를 활용한 양방향 통신을 구현하세요.

Part 1: 이동 평균 계산기

def moving_average(window_size):
    """
    이동 평균을 계산하는 코루틴
    
    사용법:
    ma = moving_average(3)
    next(ma)  # 프라이밍
    print(ma.send(10))  # 10.0
    print(ma.send(20))  # 15.0
    print(ma.send(30))  # 20.0
    print(ma.send(40))  # 30.0 (최근 3개: 20,30,40)
    """
    # yield와 send를 활용하여 구현

# 테스트 코드 작성

Part 2: 데이터 파이프라인

여러 generator를 연결하여 데이터 처리 파이프라인을 구성하세요:

def data_source():
    """데이터를 생성하는 generator"""
    import random
    while True:
        yield random.randint(1, 100)

def filter_even(target):
    """짝수만 통과시키는 필터"""
    while True:
        value = (yield)
        if value % 2 == 0:
            target.send(value)

def multiply_by(factor, target):
    """값에 factor를 곱하는 변환기"""
    while True:
        value = (yield)
        target.send(value * factor)

def sink():
    """최종 처리 - 값을 출력"""
    while True:
        value = (yield)
        print(f"Final: {value}")

# 파이프라인 구성 및 실행
# data -> filter_even -> multiply_by(2) -> sink

Part 3: 상태 기계 (State Machine)

def create_state_machine():
    """
    간단한 자판기 상태 기계
    
    상태: IDLE -> COIN_INSERTED -> ITEM_SELECTED -> DISPENSING -> IDLE
    
    명령:
    - insert_coin(amount): 동전 투입
    - select_item(item): 상품 선택
    - cancel(): 취소 및 환불
    - dispense(): 상품 배출
    
    예시:
    machine = create_state_machine()
    next(machine)
    machine.send(('insert_coin', 500))
    machine.send(('select_item', 'cola'))
    machine.send(('dispense',))
    """
    # 상태 패턴을 generator로 구현

# 테스트 케이스 작성

힌트:

def generator_example():
    total = 0
    while True:
        value = (yield total)  # 값을 받고, 현재 total을 반환
        if value is not None:
            total += value

# 여기에 코드를 작성하세요

문제 5^★★★: 함수형 프로그래밍 - Currying과 Partial Application

함수형 프로그래밍의 핵심 개념인 커링과 부분 적용을 구현하고 활용하세요.

Part 1: Curry 구현

def curry(func):
    """
    함수를 자동으로 커링하는 데코레이터
    
    예시:
    @curry
    def add(a, b, c):
        return a + b + c
    
    # 모두 동일한 결과
    add(1, 2, 3)        # 6
    add(1)(2)(3)        # 6
    add(1, 2)(3)        # 6
    add(1)(2, 3)        # 6
    """
    # 구현

# 테스트
@curry
def multiply(a, b, c, d):
    return a * b * c * d

result = multiply(2)(3)(4)(5)
print(result)  # 120

Part 2: Compose 함수

def compose(*functions):
    """
    여러 함수를 조합하는 함수
    오른쪽에서 왼쪽으로 실행
    
    예시:
    def add_one(x): return x + 1
    def double(x): return x * 2
    def square(x): return x ** 2
    
    f = compose(add_one, double, square)
    f(3)  # add_one(double(square(3))) = add_one(double(9)) = add_one(18) = 19
    """
    # 구현

# 테스트 및 활용

Part 3: Pipe 연산자

class Pipeable:
    """
    파이프라인 스타일의 함수 조합
    
    예시:
    result = (Pipeable(5)
              .pipe(lambda x: x + 1)
              .pipe(lambda x: x * 2)
              .pipe(lambda x: x ** 2)
              .value())
    """
    # 구현

# 실제 데이터 처리 예제
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

result = (Pipeable(data)
          .pipe(lambda lst: filter(lambda x: x % 2 == 0, lst))
          .pipe(lambda lst: map(lambda x: x ** 2, lst))
          .pipe(list)
          .pipe(sum)
          .value())

print(result)  # 220 (2²+4²+6²+8²+10²)

Part 4: 실전 응용 - 데이터 변환 파이프라인

# 다음 변환을 함수형 스타일로 구현하세요:
users = [
    {'name': 'Alice', 'age': 25, 'city': 'Seoul', 'score': 85},
    {'name': 'Bob', 'age': 30, 'city': 'Busan', 'score': 92},
    {'name': 'Charlie', 'age': 35, 'city': 'Seoul', 'score': 78},
    {'name': 'David', 'age': 28, 'city': 'Seoul', 'score': 95},
]

# 요구사항:
# 1. Seoul에 사는 사람만 필터링
# 2. 점수가 80점 이상인 사람만 선택
# 3. 이름과 점수만 추출
# 4. 점수 기준 내림차순 정렬

# 함수형 스타일로 구현 (compose, curry 등 활용)

# 여기에 코드를 작성하세요

문제 6^★★★★: Context Manager를 함수로 구현

contextlib를 사용하여 다양한 context manager를 generator 함수로 구현하세요.

기본 개념:

from contextlib import contextmanager

@contextmanager
def my_context():
    print("Entering context")
    yield "resource"
    print("Exiting context")

with my_context() as resource:
    print(f"Using {resource}")

구현할 Context Managers:

1. 실행 시간 측정

@contextmanager
def timer(name="Operation"):
    """
    코드 블록의 실행 시간을 측정
    
    with timer("Database query"):
        # 시간이 걸리는 작업
        time.sleep(1)
    # 출력: Database query took 1.00 seconds
    """
    # 구현

2. 임시 디렉토리 관리

@contextmanager
def temporary_directory():
    """
    임시 디렉토리를 생성하고 종료 시 자동 삭제
    
    with temporary_directory() as tmpdir:
        # tmpdir 경로를 사용
        file_path = tmpdir / "test.txt"
        file_path.write_text("test")
    # 여기서 tmpdir은 자동으로 삭제됨
    """
    # 구현

3. 데이터베이스 트랜잭션 시뮬레이션

class FakeDatabase:
    def __init__(self):
        self.data = {}
    
    def get(self, key):
        return self.data.get(key)
    
    def set(self, key, value):
        self.data[key] = value
    
    def rollback(self):
        # 롤백 로직
        pass
    
    def commit(self):
        # 커밋 로직
        pass

@contextmanager
def transaction(db):
    """
    트랜잭션 관리
    - 정상 종료: 커밋
    - 예외 발생: 롤백
    
    db = FakeDatabase()
    try:
        with transaction(db) as tx:
            tx.set('user', 'Alice')
            raise ValueError("Error!")
    except ValueError:
        pass
    
    assert db.get('user') is None  # 롤백되어 저장되지 않음
    """
    # 구현

4. 속성 임시 변경

@contextmanager
def modified_attributes(obj, **changes):
    """
    객체의 속성을 임시로 변경하고 복원
    
    class Config:
        debug = False
        timeout = 30
    
    config = Config()
    print(config.debug)  # False
    
    with modified_attributes(config, debug=True, timeout=60):
        print(config.debug)  # True
        print(config.timeout)  # 60
    
    print(config.debug)  # False (복원됨)
    """
    # 구현

5. 리소스 풀 관리

@contextmanager
def connection_pool(max_connections=5):
    """
    연결 풀을 관리하고 자동으로 정리
    
    with connection_pool(max_connections=3) as pool:
        conn1 = pool.acquire()
        conn2 = pool.acquire()
        # 사용
        pool.release(conn1)
    # 모든 연결이 자동으로 정리됨
    """
    # 구현

# 여기에 코드를 작성하세요
from contextlib import contextmanager
import time

문제 7^★★★: 메모이제이션 고급 패턴

다양한 캐싱 전략을 구현하고 성능을 비교하세요.

Part 1: LRU (Least Recently Used) 캐시

def lru_cache(maxsize=128):
    """
    LRU 캐싱 데코레이터 구현
    
    - 최근에 사용하지 않은 항목부터 제거
    - maxsize 초과 시 가장 오래된 항목 삭제
    - cache_info() 메서드로 통계 제공
    
    @lru_cache(maxsize=3)
    def fibonacci(n):
        if n < 2:
            return n
        return fibonacci(n-1) + fibonacci(n-2)
    """
    # 구현 (collections.OrderedDict 활용)

# 테스트 및 성능 비교

Part 2: TTL (Time To Live) 캐시

def ttl_cache(ttl_seconds=60):
    """
    시간 제한이 있는 캐시
    
    - 캐시된 값은 ttl_seconds 후 만료
    - 만료된 값은 자동으로 재계산
    
    @ttl_cache(ttl_seconds=5)
    def get_data():
        return expensive_operation()
    """
    # 구현 (time.time() 활용)

Part 3: 적응형 캐시

def adaptive_cache(cost_function=None):
    """
    호출 빈도와 계산 비용을 고려한 적응형 캐시
    
    - 자주 호출되고 비용이 높은 결과를 우선 캐싱
    - 캐시 히트율 모니터링
    - 동적으로 캐시 크기 조정
    
    @adaptive_cache(cost_function=lambda n: n**2)
    def expensive_func(n):
        time.sleep(0.1 * n)
        return n ** 2
    """
    # 구현

Part 4: 멀티레벨 캐시

class MultiLevelCache:
    """
    L1(메모리), L2(디스크) 캐시 시뮬레이션
    
    - L1: 빠르지만 용량 작음
    - L2: 느리지만 용량 큼
    - L1 미스 → L2 확인 → 계산
    
    사용법:
    cache = MultiLevelCache(l1_size=10, l2_size=100)
    
    @cache.cached
    def compute(x):
        return heavy_computation(x)
    """
    # 구현

Part 5: 성능 비교 및 분석

다음 함수들로 각 캐싱 전략의 성능을 비교하세요:

def benchmark_cache_strategy(cache_decorator, test_func, test_inputs):
    """
    캐싱 전략의 성능을 측정
    
    반환값:
    - 총 실행 시간
    - 캐시 히트율
    - 메모리 사용량 (추정)
    """
    # 구현

# 테스트 시나리오
def slow_fibonacci(n):
    if n < 2:
        return n
    return slow_fibonacci(n-1) + slow_fibonacci(n-2)

# 각 전략 비교
strategies = [lru_cache, ttl_cache, adaptive_cache]
test_inputs = [10, 15, 20, 10, 15, 20]  # 반복 패턴

for strategy in strategies:
    stats = benchmark_cache_strategy(strategy, slow_fibonacci, test_inputs)
    print(f"{strategy.__name__}: {stats}")

# 여기에 코드를 작성하세요

문제 8^★★★★: 동적 함수 생성 및 메타프로그래밍

런타임에 함수를 동적으로 생성하고 조작하는 고급 기법을 구현하세요.

Part 1: 함수 팩토리

def create_validator_function(rules):
    """
    검증 규칙을 받아 검증 함수를 동적으로 생성
    
    rules = {
        'type': int,
        'min': 0,
        'max': 100,
        'even': True
    }
    
    validator = create_validator_function(rules)
    validator(50)   # True
    validator(101)  # False (max 초과)
    validator(33)   # False (홀수)
    """
    # exec() 또는 compile() 활용

Part 2: DSL (Domain Specific Language) 구현

class QueryBuilder:
    """
    SQL 쿼리를 Python 함수 체이닝으로 작성
    
    예시:
    query = (QueryBuilder()
             .select('name', 'age')
             .from_table('users')
             .where(lambda row: row['age'] > 25)
             .order_by('name')
             .limit(10))
    
    results = query.execute(data)
    """
    # 구현

# 실제 데이터로 테스트
users = [
    {'name': 'Alice', 'age': 30},
    {'name': 'Bob', 'age': 25},
    {'name': 'Charlie', 'age': 35},
]

Part 3: 함수 변환기

def memoize_recursive(func):
    """
    재귀 함수를 자동으로 메모이제이션하는 변환기
    
    - 재귀 호출을 감지
    - 자동으로 캐싱 추가
    - 원본 함수 구조 유지
    """
    # 구현

@memoize_recursive
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# fibonacci(100)도 빠르게 계산되어야 함

Part 4: 함수 인스펙션 및 수정

import inspect

def analyze_function(func):
    """
    함수를 분석하여 정보 추출
    
    반환값:
    {
        'name': 함수 이름,
        'params': 파라미터 리스트,
        'defaults': 기본값,
        'annotations': 타입 힌트,
        'source': 소스 코드,
        'complexity': 순환 복잡도 (간단한 추정)
    }
    """
    # inspect 모듈 활용

def auto_document(func):
    """
    함수의 시그니처와 docstring을 자동 생성
    """
    # 구현

# 테스트
def example(x: int, y: int = 10, *args, **kwargs) -> int:
    return x + y + sum(args)

info = analyze_function(example)
print(info)

Part 5: 프로파일링 데코레이터

def profile(sort_by='cumtime', top_n=10):
    """
    함수 실행을 프로파일링하는 데코레이터
    
    - 각 라인별 실행 시간 측정
    - 메모리 사용량 추적
    - 호출 그래프 생성
    
    @profile(sort_by='time', top_n=5)
    def complex_function():
        # 복잡한 로직
        pass
    """
    # cProfile과 line_profiler 개념 활용

# 여기에 코드를 작성하세요
import inspect

문제 9^★★★★: 비동기 패턴 시뮬레이션

Generator를 활용하여 비동기 프로그래밍의 개념을 구현하세요.

배경: Python의 async/await는 generator를 기반으로 구현되었습니다. 이 개념을 이해하기 위해 간단한 버전을 만들어봅시다.

Part 1: Task와 이벤트 루프

class Task:
    """단일 비동기 작업을 나타내는 클래스"""
    def __init__(self, generator):
        self.generator = generator
        self.finished = False
        self.result = None
    
    def step(self):
        """작업을 한 단계 진행"""
        try:
            # generator의 다음 단계 실행
            return next(self.generator)
        except StopIteration as e:
            self.finished = True
            self.result = e.value
            return None

class EventLoop:
    """간단한 이벤트 루프 구현"""
    def __init__(self):
        self.tasks = []
    
    def create_task(self, generator):
        """새 작업 추가"""
        task = Task(generator)
        self.tasks.append(task)
        return task
    
    def run_until_complete(self, generator):
        """generator가 완료될 때까지 실행"""
        # 구현
        pass
    
    def run(self):
        """모든 작업이 완료될 때까지 실행"""
        while self.tasks:
            # 라운드 로빈 방식으로 각 작업 실행
            # 완료된 작업 제거
            pass

# 테스트
def task1():
    print("Task 1: Step 1")
    yield
    print("Task 1: Step 2")
    yield
    print("Task 1: Step 3")
    return "Task 1 done"

def task2():
    print("Task 2: Step 1")
    yield
    print("Task 2: Step 2")
    return "Task 2 done"

loop = EventLoop()
loop.create_task(task1())
loop.create_task(task2())
loop.run()

Part 2: Sleep 구현

def sleep(seconds):
    """
    지정된 시간만큼 대기하는 generator
    
    실제로는 다른 작업에게 실행 기회를 양보
    """
    import time
    start_time = time.time()
    while time.time() - start_time < seconds:
        yield  # 제어권 반환

# 사용 예시
def delayed_task():
    print("Starting task")
    yield from sleep(1)  # 1초 대기
    print("Task completed after 1 second")

Part 3: 비동기 HTTP 요청 시뮬레이션

class AsyncHTTPRequest:
    """비동기 HTTP 요청 시뮬레이터"""
    def __init__(self, url, delay=0.5):
        self.url = url
        self.delay = delay
    
    def fetch(self):
        """
        요청을 수행하는 generator
        실제로는 delay만큼 대기 후 결과 반환
        """
        yield from sleep(self.delay)
        return f"Response from {self.url}"

def fetch_multiple_urls(urls):
    """
    여러 URL을 동시에 가져오기
    
    순차적: N * delay 시간
    비동기: delay 시간 (병렬 실행)
    """
    results = []
    for url in urls:
        request = AsyncHTTPRequest(url)
        response = yield from request.fetch()
        results.append(response)
    return results

# 테스트
urls = ['http://api1.com', 'http://api2.com', 'http://api3.com']
loop = EventLoop()
task = loop.create_task(fetch_multiple_urls(urls))
loop.run()

Part 4: Future와 콜백

class Future:
    """미래의 결과를 나타내는 클래스"""
    def __init__(self):
        self.result = None
        self.done = False
        self.callbacks = []
    
    def set_result(self, result):
        """결과 설정 및 콜백 실행"""
        self.result = result
        self.done = True
        for callback in self.callbacks:
            callback(result)
    
    def add_callback(self, callback):
        """콜백 등록"""
        if self.done:
            callback(self.result)
        else:
            self.callbacks.append(callback)
    
    def __iter__(self):
        """Future가 완료될 때까지 대기"""
        while not self.done:
            yield
        return self.result

# 사용 예시
def async_computation():
    future = Future()
    
    def compute():
        # 시간이 걸리는 계산
        yield from sleep(1)
        future.set_result(42)
    
    # 백그라운드에서 계산 시작
    loop.create_task(compute())
    
    # 결과를 기다림
    result = yield from future
    return result

Part 5: 실전 응용 - 웹 스크래퍼

def web_scraper(urls, max_concurrent=3):
    """
    여러 URL을 동시에 스크래핑
    
    - 최대 max_concurrent개의 요청만 동시 실행
    - 각 페이지에서 링크 추출
    - 결과를 점진적으로 반환
    
    for result in web_scraper(urls):
        print(f"Scraped: {result}")
    """
    # 구현

# 여기에 코드를 작성하세요

문제 10^★★★★: 종합 프로젝트 - 플러그인 시스템

함수와 데코레이터를 활용한 확장 가능한 플러그인 시스템을 구현하세요.

요구사항:

확장 가능한 명령어 처리 시스템을 만드세요. 사용자는 쉽게 새로운 명령어를 추가할 수 있어야 합니다.

Part 1: 플러그인 레지스트리

class PluginRegistry:
    """
    플러그인을 등록하고 관리하는 시스템
    """
    def __init__(self):
        self.plugins = {}
        self.hooks = {}
    
    def register(self, name, category='default'):
        """
        플러그인 등록 데코레이터
        
        @registry.register('hello', category='greetings')
        def hello_command(name):
            return f"Hello, {name}!"
        """
        def decorator(func):
            # 구현
            pass
        return decorator
    
    def hook(self, hook_name):
        """
        훅 포인트 등록
        
        @registry.hook('before_execute')
        def log_execution(command, args):
            print(f"Executing: {command}")
        """
        def decorator(func):
            # 구현
            pass
        return decorator
    
    def execute(self, name, *args, **kwargs):
        """플러그인 실행"""
        # before 훅 실행
        # 플러그인 실행
        # after 훅 실행
        pass
    
    def list_plugins(self, category=None):
        """등록된 플러그인 목록"""
        pass

# 기본 사용
registry = PluginRegistry()

@registry.register('greet')
def greet(name):
    return f"Hello, {name}!"

@registry.register('add')
def add(a, b):
    return a + b

result = registry.execute('greet', 'Alice')
print(result)  # "Hello, Alice!"

Part 2: 명령어 파이프라인

class CommandPipeline:
    """
    명령어를 체이닝하여 실행
    
    예시:
    pipeline = (CommandPipeline()
                .add('load_data', filename='data.txt')
                .add('filter', condition=lambda x: x > 0)
                .add('transform', func=lambda x: x * 2)
                .add('save', filename='output.txt'))
    
    pipeline.execute()
    """
    def __init__(self):
        self.commands = []
    
    def add(self, command_name, **params):
        """명령어 추가"""
        # 구현
        return self
    
    def execute(self, initial_data=None):
        """파이프라인 실행"""
        # 각 명령어를 순서대로 실행
        # 이전 명령어의 출력이 다음 명령어의 입력
        pass

# 데이터 처리 명령어들 등록
@registry.register('load_data')
def load_data(filename):
    # 파일에서 데이터 로드
    pass

@registry.register('filter')
def filter_data(data, condition):
    return [x for x in data if condition(x)]

@registry.register('transform')
def transform_data(data, func):
    return [func(x) for x in data]

Part 3: 플러그인 의존성 관리

def depends_on(*dependencies):
    """
    플러그인 의존성 선언 데코레이터
    
    @depends_on('database', 'logger')
    @registry.register('save_user')
    def save_user(user_data, database, logger):
        logger.log(f"Saving user: {user_data}")
        database.save(user_data)
    """
    def decorator(func):
        # 의존성을 메타데이터로 저장
        # 실행 시 자동으로 의존성 주입
        pass
    return decorator

class DependencyInjector:
    """의존성 주입 컨테이너"""
    def __init__(self):
        self.services = {}
    
    def provide(self, name, service):
        """서비스 등록"""
        self.services[name] = service
    
    def inject(self, func):
        """함수에 의존성 자동 주입"""
        # 함수의 파라미터를 분석
        # 서비스 레지스트리에서 찾아 주입
        pass

Part 4: 플러그인 설정 및 초기화

class ConfigurablePlugin:
    """
    설정 가능한 플러그인 베이스
    """
    def __init__(self, config=None):
        self.config = config or {}
    
    def setup(self):
        """초기화 로직"""
        pass
    
    def teardown(self):
        """정리 로직"""
        pass
    
    def __call__(self, *args, **kwargs):
        """실행 로직"""
        raise NotImplementedError

# 예시: 데이터베이스 플러그인
class DatabasePlugin(ConfigurablePlugin):
    def setup(self):
        self.connection = create_connection(self.config['host'])
    
    def teardown(self):
        self.connection.close()
    
    def __call__(self, query):
        return self.connection.execute(query)

Part 5: 실전 통합

다음 기능을 가진 완전한 플러그인 시스템을 구현하세요:

# 플러그인 정의
@registry.register('weather')
@depends_on('http_client')
@cache(ttl=300)
def get_weather(city, http_client):
    """날씨 정보 조회"""
    return http_client.get(f"https://api.weather.com/{city}")

@registry.register('notify')
@depends_on('email_service', 'logger')
@retry(max_attempts=3)
def send_notification(message, email_service, logger):
    """알림 전송"""
    logger.info(f"Sending: {message}")
    email_service.send(message)

# 워크플로우 정의
workflow = (CommandPipeline()
            .add('get_weather', city='Seoul')
            .add('process_data')
            .add('notify', message='Weather update'))

# 실행
registry.execute_workflow(workflow)

테스트 케이스: 1. 플러그인 등록 및 실행 2. 의존성 주입 테스트 3. 훅 실행 순서 확인 4. 에러 처리 및 롤백 5. 플러그인 동적 로딩/언로딩

# 여기에 코드를 작성하세요