포스트

[이제와서 시작하는 Python 마스터하기 #10] 데코레이터와 제너레이터

게시
By YonYonWare
43 분읽는 시간
[이제와서 시작하는 Python 마스터하기 #10] 데코레이터와 제너레이터

🚀 실전 예제로 시작하기

📱 API 응답 캐싱 시스템

실제 웹 개발에서 자주 사용되는 API 응답 캐싱 시스템을 데코레이터로 구현해보겠습니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

import time
import hashlib
import json
import requests
from functools import wraps

# 한국 공공데이터 API 시뮬레이션
class KoreanDataAPI:
    """한국 공공데이터 API 클라이언트"""

    def __init__(self):
        self.cache = {}  # 실제로는 Redis 등 사용
        self.api_calls = 0

    def cache_response(self, expire_time=300):  # 5분 캐시
        """API 응답 캐싱 데코레이터"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 캐시 키 생성
                cache_key = self._generate_cache_key(func.__name__, args, kwargs)

                # 캐시 확인
                if cache_key in self.cache:
                    cached_data, timestamp = self.cache[cache_key]
                    if time.time() - timestamp < expire_time:
                        print(f"💾 캐시에서 데이터 반환: {func.__name__}")
                        return cached_data

                # API 호출
                print(f"🌐 API 호출: {func.__name__}")
                self.api_calls += 1
                result = func(*args, **kwargs)

                # 캐시 저장
                self.cache[cache_key] = (result, time.time())

                return result
            return wrapper
        return decorator

    def _generate_cache_key(self, func_name, args, kwargs):
        """캐시 키 생성"""
        key_data = f"{func_name}:{args}:{kwargs}"
        return hashlib.md5(key_data.encode()).hexdigest()

    @cache_response(expire_time=600)  # 10분 캐시
    def get_weather_data(self, city="서울"):
        """날씨 데이터 조회"""
        # 실제 API 호출 시뮬레이션
        time.sleep(1)  # 네트워크 지연 시뮬레이션

        weather_data = {
            "city": city,
            "temperature": 23,
            "humidity": 65,
            "description": "맑음",
            "timestamp": time.time()
        }
        return weather_data

    @cache_response(expire_time=1800)  # 30분 캐시
    def get_population_data(self, region="전국"):
        """인구 데이터 조회"""
        time.sleep(0.5)

        population_data = {
            "region": region,
            "population": 51_780_000,
            "growth_rate": 0.1,
            "last_updated": "2024년 1월"
        }
        return population_data

# 사용 예제
api = KoreanDataAPI()

# 첫 번째 호출 - API 호출
weather1 = api.get_weather_data("서울")
print(f"날씨 데이터: {weather1}")

# 두 번째 호출 - 캐시에서 반환
weather2 = api.get_weather_data("서울")
print(f"API 호출 횟수: {api.api_calls}")

📊 대용량 데이터 스트리밍 처리

제너레이터를 활용한 대용량 CSV 파일 처리 시스템입니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

import csv
import json
from typing import Iterator, Dict, Any

class KoreanLogAnalyzer:
    """한국 웹 서비스 로그 분석기"""

    def __init__(self, log_file_path: str):
        self.log_file_path = log_file_path
        self.stats = {
            "total_requests": 0,
            "unique_users": set(),
            "error_count": 0,
            "peak_hour": {},
            "popular_pages": {}
        }

    def read_log_file(self) -> Iterator[Dict[str, str]]:
        """로그 파일을 한 줄씩 읽는 제너레이터"""
        with open(self.log_file_path, 'r', encoding='utf-8') as file:
            reader = csv.DictReader(file)
            for row in reader:
                yield row

    def filter_korean_users(self, log_stream: Iterator[Dict]) -> Iterator[Dict]:
        """한국 사용자만 필터링"""
        for log in log_stream:
            # IP 기반 또는 User-Agent 기반 한국 사용자 필터링
            if (log.get('country') == 'KR' or
                'ko-kr' in log.get('user_agent', '').lower()):
                yield log

    def extract_business_hours(self, log_stream: Iterator[Dict]) -> Iterator[Dict]:
        """업무시간(9-18시) 데이터만 추출"""
        for log in log_stream:
            hour = int(log.get('timestamp', '00:00:00').split(':')[0])
            if 9 <= hour <= 18:
                yield log

    def analyze_user_behavior(self, log_stream: Iterator[Dict]) -> Iterator[Dict]:
        """사용자 행동 분석 데이터 생성"""
        for log in log_stream:
            self.stats["total_requests"] += 1
            self.stats["unique_users"].add(log.get('user_id'))

            if log.get('status_code', '200').startswith('4'):
                self.stats["error_count"] += 1

            # 시간대별 통계
            hour = log.get('timestamp', '00:00:00')[:2]
            self.stats["peak_hour"][hour] = self.stats["peak_hour"].get(hour, 0) + 1

            # 인기 페이지 통계
            page = log.get('page', '')
            self.stats["popular_pages"][page] = self.stats["popular_pages"].get(page, 0) + 1

            # 분석된 데이터 반환
            yield {
                "user_id": log.get('user_id'),
                "page": log.get('page'),
                "session_time": log.get('session_time'),
                "conversion": log.get('status_code') == '200'
            }

    def process_logs(self, batch_size: int = 1000) -> Iterator[List[Dict]]:
        """로그 데이터를 배치 단위로 처리"""
        # 데이터 파이프라인 구성
        log_stream = self.read_log_file()
        korean_users = self.filter_korean_users(log_stream)
        business_hours = self.extract_business_hours(korean_users)
        analyzed_data = self.analyze_user_behavior(business_hours)

        # 배치 처리
        batch = []
        for data in analyzed_data:
            batch.append(data)

            if len(batch) >= batch_size:
                yield batch
                batch = []

        # 마지막 배치
        if batch:
            yield batch

    def generate_daily_report(self):
        """일일 리포트 생성"""
        print("📊 한국 사용자 웹 로그 분석 리포트")
        print(f"전체 요청 수: {self.stats['total_requests']:,}")
        print(f"고유 사용자 수: {len(self.stats['unique_users']):,}")
        print(f"에러율: {(self.stats['error_count']/self.stats['total_requests']*100):.2f}%")

        # 피크 시간 찾기
        peak_hour = max(self.stats['peak_hour'].items(), key=lambda x: x[1])
        print(f"피크 시간: {peak_hour[0]}시 ({peak_hour[1]:,}건)")

        # 인기 페이지 TOP 3
        popular = sorted(self.stats['popular_pages'].items(),
                        key=lambda x: x[1], reverse=True)[:3]
        print("인기 페이지 TOP 3:")
        for i, (page, count) in enumerate(popular, 1):
            print(f"  {i}. {page}: {count:,}회")

# 사용 예제 (CSV 파일이 있다고 가정)
def create_sample_log_file():
    """샘플 로그 파일 생성"""
    import random

    sample_data = []
    pages = ['/home', '/products', '/cart', '/checkout', '/login']
    countries = ['KR', 'US', 'JP', 'CN']

    for i in range(10000):
        sample_data.append({
            'user_id': f'user_{random.randint(1, 1000)}',
            'timestamp': f'{random.randint(0, 23):02d}:{random.randint(0, 59):02d}:{random.randint(0, 59):02d}',
            'page': random.choice(pages),
            'status_code': random.choice(['200', '200', '200', '404', '500']),
            'country': random.choice(countries),
            'user_agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; ko-kr)',
            'session_time': random.randint(10, 300)
        })

    with open('sample_logs.csv', 'w', newline='', encoding='utf-8') as file:
        fieldnames = ['user_id', 'timestamp', 'page', 'status_code', 'country', 'user_agent', 'session_time']
        writer = csv.DictWriter(file, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(sample_data)

# 실행
create_sample_log_file()
analyzer = KoreanLogAnalyzer('sample_logs.csv')

# 배치 처리로 메모리 효율적 분석
for batch_num, batch in enumerate(analyzer.process_logs(batch_size=500), 1):
    print(f"배치 {batch_num} 처리 완료: {len(batch)}건")
    if batch_num >= 5:  # 처음 5개 배치만 출력
        break

# 리포트 생성
analyzer.generate_daily_report()

🎨 데코레이터(Decorator)란?

데코레이터는 함수나 클래스를 수정하지 않고 기능을 추가하는 Python의 강력한 기능입니다.

graph LR
    A[원본 함수] --> B[데코레이터]
    B --> C[확장된 함수]
    
    D[함수 입력] --> C
    C --> E[전처리]
    E --> F[원본 함수 실행]
    F --> G[후처리]
    G --> H[결과 반환]

데코레이터 기초

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 간단한 데코레이터
def my_decorator(func):
    """기본 데코레이터"""
    def wrapper():
        print("함수 실행 전")
        result = func()
        print("함수 실행 후")
        return result
    return wrapper

# 데코레이터 사용 방법 1: @ 문법
@my_decorator
def say_hello():
    print("안녕하세요!")

say_hello()
# 출력:
# 함수 실행 전
# 안녕하세요!
# 함수 실행 후

# 데코레이터 사용 방법 2: 수동 적용
def say_goodbye():
    print("안녕히 가세요!")

decorated_goodbye = my_decorator(say_goodbye)
decorated_goodbye()

인자가 있는 함수 데코레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import functools

def universal_decorator(func):
    """모든 함수에 사용 가능한 데코레이터"""
    @functools.wraps(func)  # 원본 함수의 메타데이터 보존
    def wrapper(*args, **kwargs):
        print(f"함수 '{func.__name__}' 호출")
        print(f"위치 인자: {args}")
        print(f"키워드 인자: {kwargs}")
        result = func(*args, **kwargs)
        print(f"반환값: {result}")
        return result
    return wrapper

@universal_decorator
def add(a, b):
    """두 수를 더하는 함수"""
    return a + b

@universal_decorator
def greet(name, greeting="안녕"):
    """인사하는 함수"""
    return f"{greeting}, {name}님!"

# 사용
result = add(5, 3)
message = greet("철수", greeting="반가워")

매개변수가 있는 데코레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

def repeat(times):
    """함수를 여러 번 실행하는 데코레이터"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for i in range(times):
                print(f"{i+1}번째 실행:")
                result = func(*args, **kwargs)
                results.append(result)
            return results
        return wrapper
    return decorator

@repeat(times=3)
def greet(name):
    return f"안녕하세요, {name}님!"

results = greet("파이썬")
print(f"모든 결과: {results}")

# 실행 시간 측정 데코레이터
import time

def timeit(func):
    """함수 실행 시간을 측정하는 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 실행 시간: {end - start:.4f}초")
        return result
    return wrapper

@timeit
def slow_function():
    """시간이 걸리는 함수"""
    time.sleep(1)
    return "완료!"

result = slow_function()

클래스 데코레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# 함수를 데코레이트하는 클래스
class CountCalls:
    """함수 호출 횟수를 세는 데코레이터"""
    
    def __init__(self, func):
        self.func = func
        self.count = 0
        functools.update_wrapper(self, func)
    
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"{self.func.__name__} 호출 횟수: {self.count}")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello():
    print("Hello!")

say_hello()  # 호출 횟수: 1
say_hello()  # 호출 횟수: 2
say_hello()  # 호출 횟수: 3

# 클래스를 데코레이트하는 데코레이터
def add_repr(cls):
    """클래스에 __repr__ 메서드를 추가하는 데코레이터"""
    def __repr__(self):
        attrs = ', '.join(f"{k}={v!r}" for k, v in self.__dict__.items())
        return f"{cls.__name__}({attrs})"
    
    cls.__repr__ = __repr__
    return cls

@add_repr
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

person = Person("김철수", 30)
print(person)  # Person(name='김철수', age=30)

다중 데코레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

def bold(func):
    """굵은 글씨 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return f"<b>{result}</b>"
    return wrapper

def italic(func):
    """기울임 글씨 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return f"<i>{result}</i>"
    return wrapper

def underline(func):
    """밑줄 데코레이터"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        return f"<u>{result}</u>"
    return wrapper

# 다중 데코레이터 적용 (아래에서 위로 적용)
@bold
@italic
@underline
def say(text):
    return text

result = say("Hello, Python!")
print(result)  # <b><i><u>Hello, Python!</u></i></b>

🔄 제너레이터(Generator)

제너레이터는 메모리 효율적으로 시퀀스를 생성하는 특별한 함수입니다.

제너레이터 기초

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# 일반 함수 vs 제너레이터
def normal_function():
    """모든 값을 메모리에 저장"""
    result = []
    for i in range(1000000):
        result.append(i ** 2)
    return result

def generator_function():
    """필요할 때마다 값을 생성"""
    for i in range(1000000):
        yield i ** 2

> [!TIP]
> **yield는 "잠깐 멈춤"입니다!**
>
> `return`은 값을 반환하고 함수를 **완전히 끝내버리지만**,
> `yield`는 값을 던져주고 **잠시 멈춰있다가**, 다음 호출이 오면 **멈춘 곳에서 다시 시작**합니다.
> 그래서 메모리를 적게 쓰면서 엄청나게 많은 데이터를 처리할 수 있는 거죠!
# 메모리 사용량 비교
import sys

# 리스트: 모든 값을 메모리에 저장
numbers_list = normal_function()
print(f"리스트 크기: {sys.getsizeof(numbers_list):,} bytes")

# 제너레이터: 값을 필요할 때 생성
numbers_gen = generator_function()
print(f"제너레이터 크기: {sys.getsizeof(numbers_gen):,} bytes")

# 제너레이터 사용
gen = generator_function()
print(next(gen))  # 0
print(next(gen))  # 1
print(next(gen))  # 4

# for 루프에서 사용
for i, value in enumerate(generator_function()):
    if i >= 5:
        break
    print(value, end=' ')  # 0 1 4 9 16

[!WARNING] 제너레이터는 일회용입니다!

제너레이터는 한 번 끝까지 순회하면(소진되면) 다시 사용할 수 없습니다. 다시 쓰려면 제너레이터 객체를 새로 생성해야 합니다. list(gen) 처럼 리스트로 변환해두면 여러 번 쓸 수 있지만, 메모리를 많이 쓰게 되니 주의하세요!

yield 키워드

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

def fibonacci_generator(n):
    """피보나치 수열 제너레이터"""
    a, b = 0, 1
    count = 0
    
    while count < n:
        yield a
        a, b = b, a + b
        count += 1

# 사용
fib_gen = fibonacci_generator(10)
for num in fib_gen:
    print(num, end=' ')  # 0 1 1 2 3 5 8 13 21 34

# 무한 제너레이터
def infinite_counter(start=0):
    """무한 카운터"""
    while True:
        yield start
        start += 1

# 사용 (주의: 무한 루프)
counter = infinite_counter(10)
for _ in range(5):
    print(next(counter), end=' ')  # 10 11 12 13 14

# yield from (Python 3.3+)
def flatten(nested_list):
    """중첩 리스트 평탄화"""
    for item in nested_list:
        if isinstance(item, list):
            yield from flatten(item)  # 재귀적으로 평탄화
        else:
            yield item

nested = [1, [2, 3, [4, 5]], 6, [7, [8, 9]]]
print(list(flatten(nested)))  # [1, 2, 3, 4, 5, 6, 7, 8, 9]

제너레이터 표현식

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

# 리스트 컴프리헨션 vs 제너레이터 표현식
# 리스트 컴프리헨션 (대괄호)
squares_list = [x**2 for x in range(10)]
print(squares_list)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# 제너레이터 표현식 (소괄호)
squares_gen = (x**2 for x in range(10))
print(squares_gen)  # <generator object <genexpr> at ...>
print(list(squares_gen))  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# 메모리 효율적인 처리
# 큰 파일 읽기
def read_large_file(file_path):
    """대용량 파일을 한 줄씩 읽기"""
    with open(file_path, 'r') as file:
        for line in file:
            yield line.strip()

# 사용 예
# for line in read_large_file("large_file.txt"):
#     process_line(line)

# 제너레이터 체이닝
def numbers():
    """숫자 생성"""
    for i in range(10):
        yield i

def squares(nums):
    """제곱 계산"""
    for n in nums:
        yield n ** 2

def add_one(nums):
    """1 더하기"""
    for n in nums:
        yield n + 1

# 체이닝
nums = numbers()
squared = squares(nums)
result = add_one(squared)

print(list(result))  # [1, 2, 5, 10, 17, 26, 37, 50, 65, 82]

제너레이터와 상태 관리

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

def stateful_generator():
    """상태를 가진 제너레이터"""
    total = 0
    count = 0
    
    while True:
        value = yield total
        if value is not None:
            total += value
            count += 1
            print(f"받은 값: {value}, 총합: {total}, 개수: {count}")

# 사용
gen = stateful_generator()
next(gen)  # 제너레이터 시작

gen.send(10)  # 받은 값: 10, 총합: 10, 개수: 1
gen.send(20)  # 받은 값: 20, 총합: 30, 개수: 2
gen.send(30)  # 받은 값: 30, 총합: 60, 개수: 3

# 제너레이터 종료
gen.close()

💡 실전 예제

1. 고급 데코레이터 패턴

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

import functools
import time
from typing import Any, Callable, TypeVar, cast
import asyncio

# 캐싱 데코레이터
def memoize(func: Callable) -> Callable:
    """결과를 캐싱하는 데코레이터"""
    cache = {}
    
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # 캐시 키 생성
        key = str(args) + str(kwargs)
        
        if key not in cache:
            cache[key] = func(*args, **kwargs)
        
        return cache[key]
    
    wrapper.cache = cache  # 캐시 접근 가능
    wrapper.clear_cache = lambda: cache.clear()  # 캐시 초기화
    
    return wrapper

@memoize
def expensive_function(n):
    """비용이 많이 드는 계산"""
    print(f"계산 중... {n}")
    time.sleep(1)
    return n ** 2

# 사용
print(expensive_function(5))  # 계산 중... 5 (1초 대기)
print(expensive_function(5))  # 캐시에서 즉시 반환
print(expensive_function.cache)  # 캐시 내용 확인

# 재시도 데코레이터
def retry(max_attempts=3, delay=1, exceptions=(Exception,)):
    """실패 시 재시도하는 데코레이터"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"시도 {attempt + 1} 실패: {e}")
                    time.sleep(delay)
            return None
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5, exceptions=(ValueError, ConnectionError))
def unreliable_api_call():
    """불안정한 API 호출 시뮬레이션"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("연결 실패")
    return "성공!"

# 속도 제한 데코레이터
from collections import deque
from datetime import datetime, timedelta

def rate_limit(calls=10, period=60):
    """API 호출 속도 제한 데코레이터"""
    def decorator(func):
        calls_times = deque()
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            now = datetime.now()
            
            # 오래된 호출 기록 제거
            while calls_times and calls_times[0] < now - timedelta(seconds=period):
                calls_times.popleft()
            
            if len(calls_times) >= calls:
                raise Exception(f"{period}초 동안 {calls}회 호출 제한 초과")
            
            calls_times.append(now)
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

@rate_limit(calls=5, period=10)
def api_call(endpoint):
    """API 호출"""
    return f"Called {endpoint}"

# 권한 확인 데코레이터
class User:
    def __init__(self, name, role):
        self.name = name
        self.role = role

current_user = None

def require_role(role):
    """특정 역할이 필요한 함수 데코레이터"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if not current_user:
                raise PermissionError("로그인이 필요합니다")
            if current_user.role != role:
                raise PermissionError(f"{role} 권한이 필요합니다")
            return func(*args, **kwargs)
        return wrapper
    return decorator

@require_role("admin")
def delete_user(user_id):
    """사용자 삭제 (관리자만)"""
    return f"사용자 {user_id} 삭제됨"

# 테스트
current_user = User("일반사용자", "user")
try:
    delete_user(123)
except PermissionError as e:
    print(f"권한 오류: {e}")

current_user = User("관리자", "admin")
print(delete_user(123))

2. 데이터 파이프라인 제너레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

import csv
import json
from typing import Iterator, Dict, Any, List
from datetime import datetime

# 데이터 읽기 제너레이터
def read_csv_lazy(filepath: str) -> Iterator[Dict[str, str]]:
    """CSV 파일을 한 줄씩 읽는 제너레이터"""
    with open(filepath, 'r', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            yield row

def read_json_lines(filepath: str) -> Iterator[Dict[str, Any]]:
    """JSON Lines 파일을 한 줄씩 읽는 제너레이터"""
    with open(filepath, 'r', encoding='utf-8') as file:
        for line in file:
            if line.strip():
                yield json.loads(line)

# 데이터 변환 제너레이터
def filter_data(data_stream: Iterator[Dict], condition: Callable) -> Iterator[Dict]:
    """조건에 맞는 데이터만 통과시키는 필터"""
    for item in data_stream:
        if condition(item):
            yield item

def transform_data(data_stream: Iterator[Dict], transformer: Callable) -> Iterator[Dict]:
    """데이터를 변환하는 제너레이터"""
    for item in data_stream:
        yield transformer(item)

def batch_data(data_stream: Iterator[Any], batch_size: int) -> Iterator[List[Any]]:
    """데이터를 배치로 묶는 제너레이터"""
    batch = []
    for item in data_stream:
        batch.append(item)
        if len(batch) >= batch_size:
            yield batch
            batch = []
    
    # 마지막 배치
    if batch:
        yield batch

# 파이프라인 예제
class DataPipeline:
    """데이터 처리 파이프라인"""
    
    def __init__(self):
        self.stats = {
            'total_processed': 0,
            'filtered_out': 0,
            'errors': 0
        }
    
    def process_sales_data(self, filepath: str) -> Iterator[Dict]:
        """판매 데이터 처리 파이프라인"""
        # 1. 데이터 읽기
        raw_data = read_csv_lazy(filepath)
        
        # 2. 데이터 검증 및 변환
        for row in raw_data:
            self.stats['total_processed'] += 1
            
            try:
                # 날짜 파싱
                row['date'] = datetime.strptime(row.get('date', ''), '%Y-%m-%d')
                
                # 숫자 변환
                row['amount'] = float(row.get('amount', 0))
                row['quantity'] = int(row.get('quantity', 0))
                
                # 유효성 검사
                if row['amount'] <= 0 or row['quantity'] <= 0:
                    self.stats['filtered_out'] += 1
                    continue
                
                yield row
                
            except (ValueError, KeyError) as e:
                self.stats['errors'] += 1
                print(f"데이터 처리 오류: {e}")
    
    def aggregate_by_date(self, data_stream: Iterator[Dict]) -> Iterator[Dict]:
        """날짜별로 집계"""
        daily_totals = {}
        
        for item in data_stream:
            date = item['date'].date()
            if date not in daily_totals:
                daily_totals[date] = {
                    'date': date,
                    'total_amount': 0,
                    'total_quantity': 0,
                    'transaction_count': 0
                }
            
            daily_totals[date]['total_amount'] += item['amount']
            daily_totals[date]['total_quantity'] += item['quantity']
            daily_totals[date]['transaction_count'] += 1
        
        # 정렬된 결과 반환
        for date in sorted(daily_totals.keys()):
            yield daily_totals[date]

# 무한 스트림 제너레이터
def sensor_data_stream():
    """센서 데이터 스트림 시뮬레이션"""
    import random
    import time
    
    sensor_id = "SENSOR-001"
    
    while True:
        data = {
            'sensor_id': sensor_id,
            'timestamp': datetime.now(),
            'temperature': round(20 + random.uniform(-5, 5), 2),
            'humidity': round(50 + random.uniform(-10, 10), 2),
            'pressure': round(1013 + random.uniform(-20, 20), 2)
        }
        yield data
        time.sleep(0.1)  # 100ms 간격

def moving_average(data_stream: Iterator[Dict], window_size: int, field: str):
    """이동 평균 계산 제너레이터"""
    window = deque(maxlen=window_size)
    
    for item in data_stream:
        value = item.get(field)
        if value is not None:
            window.append(value)
            
            if len(window) == window_size:
                avg = sum(window) / len(window)
                item[f'{field}_avg'] = round(avg, 2)
            
            yield item

# 사용 예제
def demo_pipeline():
    """파이프라인 데모"""
    # 센서 데이터 스트림 처리
    sensor_stream = sensor_data_stream()
    
    # 온도 이동 평균 계산
    averaged_stream = moving_average(sensor_stream, window_size=5, field='temperature')
    
    # 배치 처리
    batched_stream = batch_data(averaged_stream, batch_size=10)
    
    # 첫 번째 배치만 처리
    first_batch = next(batched_stream)
    
    print("센서 데이터 배치:")
    for data in first_batch:
        if 'temperature_avg' in data:
            print(f"시간: {data['timestamp'].strftime('%H:%M:%S')}, "
                  f"온도: {data['temperature']}°C, "
                  f"이동평균: {data['temperature_avg']}°C")

3. 컨텍스트 관리자와 데코레이터

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

from contextlib import contextmanager
import sqlite3
import os
import tempfile

# 컨텍스트 관리자 데코레이터
@contextmanager
def temporary_file(suffix=""):
    """임시 파일 생성 컨텍스트 관리자"""
    # 임시 파일 생성
    fd, path = tempfile.mkstemp(suffix=suffix)
    try:
        # 파일 디스크립터를 파일 객체로 변환
        with os.fdopen(fd, 'w') as file:
            yield file, path
    finally:
        # 정리 작업
        if os.path.exists(path):
            os.remove(path)

# 사용
with temporary_file(suffix=".txt") as (file, path):
    file.write("임시 데이터")
    print(f"임시 파일 경로: {path}")
# 파일이 자동으로 삭제됨

# 데이터베이스 트랜잭션 관리자
@contextmanager
def database_transaction(db_path):
    """데이터베이스 트랜잭션 관리"""
    conn = sqlite3.connect(db_path)
    try:
        yield conn
        conn.commit()  # 성공 시 커밋
    except Exception as e:
        conn.rollback()  # 실패 시 롤백
        raise e
    finally:
        conn.close()

# 성능 측정 컨텍스트 관리자
@contextmanager
def timer_context(name="작업"):
    """실행 시간 측정 컨텍스트 관리자"""
    import time
    start = time.time()
    print(f"{name} 시작...")
    try:
        yield
    finally:
        end = time.time()
        print(f"{name} 완료: {end - start:.4f}초")

# 리소스 풀 관리
class ResourcePool:
    """리소스 풀 관리 클래스"""
    
    def __init__(self, resource_factory, max_size=10):
        self.resource_factory = resource_factory
        self.max_size = max_size
        self.pool = []
        self.in_use = set()
    
    @contextmanager
    def get_resource(self):
        """리소스 획득 및 반환"""
        resource = None
        
        try:
            # 풀에서 리소스 가져오기
            if self.pool:
                resource = self.pool.pop()
            elif len(self.in_use) < self.max_size:
                resource = self.resource_factory()
            else:
                raise Exception("리소스 풀이 가득 찼습니다")
            
            self.in_use.add(resource)
            yield resource
            
        finally:
            # 리소스 반환
            if resource is not None:
                self.in_use.discard(resource)
                self.pool.append(resource)

# 데코레이터와 제너레이터 결합
def generate_with_cleanup(cleanup_func):
    """정리 작업이 있는 제너레이터 데코레이터"""
    def decorator(gen_func):
        @functools.wraps(gen_func)
        def wrapper(*args, **kwargs):
            gen = gen_func(*args, **kwargs)
            try:
                yield from gen
            finally:
                cleanup_func()
        return wrapper
    return decorator

# 사용 예제
def cleanup():
    print("정리 작업 수행")

@generate_with_cleanup(cleanup)
def data_generator():
    """데이터 생성 제너레이터"""
    for i in range(5):
        yield i * 2

# 프로퍼티 데코레이터 활용
class Temperature:
    """온도 변환 클래스"""
    
    def __init__(self, celsius=0):
        self._celsius = celsius
    
    @property
    def celsius(self):
        return self._celsius
    
    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError("절대영도보다 낮을 수 없습니다")
        self._celsius = value
    
    @property
    def fahrenheit(self):
        return self._celsius * 9/5 + 32
    
    @fahrenheit.setter
    def fahrenheit(self, value):
        self.celsius = (value - 32) * 5/9
    
    @property
    def kelvin(self):
        return self._celsius + 273.15
    
    @kelvin.setter
    def kelvin(self, value):
        self.celsius = value - 273.15

# 사용
temp = Temperature()
temp.celsius = 25
print(f"섭씨: {temp.celsius}°C")
print(f"화씨: {temp.fahrenheit}°F")
print(f"켈빈: {temp.kelvin}K")

⚠️ 초보자가 자주 하는 실수들

1. 데코레이터 순서 실수

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# ❌ 잘못된 데코레이터 순서
def log_calls(func):
    def wrapper(*args, **kwargs):
        print(f"호출: {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

def validate_positive(func):
    def wrapper(x):
        if x < 0:
            raise ValueError("양수여야 합니다")
        return func(x)
    return wrapper

# 잘못된 순서 - validate_positive가 먼저 실행되어야 함
@log_calls
@validate_positive  # 이게 먼저 적용됨
def calculate_sqrt(x):
    return x ** 0.5

# ✅ 올바른 순서
@validate_positive  # 먼저 검증
@log_calls         # 그 다음 로깅
def calculate_sqrt_correct(x):
    return x ** 0.5

# 테스트
try:
    calculate_sqrt(-1)  # 검증 전에 로깅이 일어남
except ValueError as e:
    print(f"에러: {e}")

try:
    calculate_sqrt_correct(-1)  # 검증 후 에러, 로깅 안됨
except ValueError as e:
    print(f"에러: {e}")

2. functools.wraps 누락

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# ❌ functools.wraps 없이 데코레이터 작성
def my_decorator_bad(func):
    def wrapper(*args, **kwargs):
        print("실행 전")
        result = func(*args, **kwargs)
        print("실행 후")
        return result
    return wrapper

@my_decorator_bad
def greet(name):
    """인사하는 함수"""
    return f"안녕하세요, {name}님!"

print(f"함수 이름: {greet.__name__}")  # wrapper
print(f"함수 문서: {greet.__doc__}")   # None

# ✅ functools.wraps 사용
from functools import wraps

def my_decorator_good(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("실행 전")
        result = func(*args, **kwargs)
        print("실행 후")
        return result
    return wrapper

@my_decorator_good
def greet_good(name):
    """인사하는 함수"""
    return f"안녕하세요, {name}님!"

print(f"함수 이름: {greet_good.__name__}")  # greet_good
print(f"함수 문서: {greet_good.__doc__}")   # 인사하는 함수

3. 제너레이터 메모리 이슈 오해

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# ❌ 제너레이터를 여러 번 사용하려는 실수
def number_generator():
    for i in range(5):
        print(f"생성: {i}")
        yield i

gen = number_generator()

# 첫 번째 사용
print("첫 번째 반복:")
for num in gen:
    print(num)

# 두 번째 사용 - 아무것도 출력되지 않음!
print("두 번째 반복:")
for num in gen:  # 제너레이터가 이미 소진됨
    print(num)

# ✅ 올바른 방법 - 함수를 호출해서 새 제너레이터 생성
print("올바른 방법:")
gen1 = number_generator()
for num in gen1:
    print(f"첫 번째: {num}")

gen2 = number_generator()
for num in gen2:
    print(f"두 번째: {num}")

4. 제너레이터와 리스트 컴프리헨션 혼동

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# ❌ 메모리 사용량을 잘못 이해하는 경우
import sys

# 큰 리스트 - 메모리를 많이 사용
big_list = [x**2 for x in range(1000000)]
print(f"리스트 메모리: {sys.getsizeof(big_list):,} bytes")

# 제너레이터 - 메모리 효율적
big_gen = (x**2 for x in range(1000000))
print(f"제너레이터 메모리: {sys.getsizeof(big_gen):,} bytes")

# ❌ 제너레이터에서 len() 사용하려는 실수
try:
    print(len(big_gen))  # TypeError 발생
except TypeError as e:
    print(f"에러: {e}")

# ✅ 올바른 방법 - 필요한 경우만 리스트로 변환
big_gen_new = (x**2 for x in range(10))  # 작은 예제
result_list = list(big_gen_new)
print(f"변환된 리스트 길이: {len(result_list)}")

5. 무한 제너레이터 잘못 사용하기

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

# ❌ 무한 제너레이터를 list()로 변환하려는 실수
def infinite_counter():
    count = 0
    while True:
        yield count
        count += 1

counter = infinite_counter()

# 이렇게 하면 메모리 부족으로 프로그램이 멈춤!
# result = list(counter)  # 절대 하지 마세요!

# ✅ 올바른 방법 - 필요한 만큼만 사용
counter = infinite_counter()
for i, value in enumerate(counter):
    print(value)
    if i >= 10:  # 10개만 출력하고 중단
        break

# 또는 islice 사용
from itertools import islice

counter = infinite_counter()
first_ten = list(islice(counter, 10))
print(f"처음 10개: {first_ten}")

6. 클로저와 데코레이터 변수 스코프 실수

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# ❌ 클로저에서 변수 스코프 실수
def create_multipliers():
    multipliers = []
    for i in range(3):
        # 잘못된 방법 - 모든 함수가 같은 i 값을 참조
        multipliers.append(lambda x: x * i)
    return multipliers

funcs = create_multipliers()
for func in funcs:
    print(func(10))  # 20, 20, 20 (모두 마지막 i 값인 2를 사용)

# ✅ 올바른 방법 - 기본 매개변수로 값 고정
def create_multipliers_correct():
    multipliers = []
    for i in range(3):
        # 기본 매개변수로 현재 i 값 캡처
        multipliers.append(lambda x, mult=i: x * mult)
    return multipliers

funcs_correct = create_multipliers_correct()
for func in funcs_correct:
    print(func(10))  # 0, 10, 20

7. 데코레이터에서 self 누락

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

# ❌ 메서드 데코레이터에서 self를 고려하지 않는 실수
def timer_bad(func):
    def wrapper(*args, **kwargs):
        import time
        start = time.time()
        result = func(*args, **kwargs)  # self가 포함된 args
        end = time.time()
        print(f"{func.__name__} 실행 시간: {end-start:.4f}초")
        return result
    return wrapper

class Calculator:
    @timer_bad
    def add(self, a, b):
        return a + b

    @timer_bad
    def slow_operation(self):
        import time
        time.sleep(0.1)
        return "완료"

# 이것은 동작하지만, self를 명시적으로 처리하는 것이 좋음
calc = Calculator()
result = calc.add(5, 3)

# ✅ 올바른 방법 - 메서드임을 명확히 하는 데코레이터
def method_timer(func):
    """메서드 전용 타이머 데코레이터"""
    @wraps(func)
    def wrapper(self, *args, **kwargs):
        import time
        start = time.time()
        result = func(self, *args, **kwargs)
        end = time.time()
        class_name = self.__class__.__name__
        print(f"{class_name}.{func.__name__} 실행 시간: {end-start:.4f}초")
        return result
    return wrapper

class BetterCalculator:
    @method_timer
    def add(self, a, b):
        return a + b

    @method_timer
    def multiply(self, a, b):
        import time
        time.sleep(0.05)  # 시뮬레이션
        return a * b

calc2 = BetterCalculator()
calc2.add(10, 20)
calc2.multiply(5, 6)

💡 실수 방지 체크리스트

✅ 데코레이터 작성 시:

  • @functools.wraps(func) 사용했는가?
  • 데코레이터 순서를 올바르게 정했는가?
  • 메서드 데코레이터에서 self를 고려했는가?

✅ 제너레이터 사용 시:

  • 제너레이터는 일회용임을 기억하는가?
  • 무한 제너레이터를 list()로 변환하지 않는가?
  • 메모리 효율성이 정말 필요한 상황인가?

✅ 클로저 사용 시:

  • 변수 스코프를 올바르게 이해하고 있는가?
  • 반복문에서 람다를 생성할 때 기본 매개변수를 사용하는가?

🎯 핵심 정리

데코레이터 Best Practices

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

# 1. functools.wraps 사용
import functools

def my_decorator(func):
    @functools.wraps(func)  # 원본 함수의 메타데이터 보존
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# 2. 클래스 기반 데코레이터
class Decorator:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
    
    def __call__(self, *args, **kwargs):
        return self.func(*args, **kwargs)

# 3. 데코레이터 팩토리
def decorator_factory(param):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # param 사용
            return func(*args, **kwargs)
        return wrapper
    return decorator

제너레이터 Best Practices

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 1. 메모리 효율적인 처리
def process_large_file(filename):
    with open(filename) as f:
        for line in f:  # 한 줄씩 읽기
            yield process_line(line)

# 2. 무한 시퀀스
def fibonacci():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# 3. 제너레이터 표현식 활용
total = sum(x**2 for x in range(1000000))  # 메모리 효율적

# 4. yield from 사용
def chain(*iterables):
    for iterable in iterables:
        yield from iterable

언제 사용할까?

graph TD
    A[데코레이터 사용 시기] --> B[횡단 관심사<br/>로깅, 캐싱, 권한]
    A --> C[함수 동작 수정<br/>재시도, 검증]
    A --> D[메타프로그래밍]
    
    E[제너레이터 사용 시기] --> F[대용량 데이터<br/>스트리밍]
    E --> G[무한 시퀀스]
    E --> H[지연 평가]
    E --> I[메모리 절약]

🎓 파이썬 마스터하기 시리즈

📚 기초편 (1-7)

  1. Python 소개와 개발 환경 설정
  2. 변수, 자료형, 연산자 완벽 정리
  3. 조건문과 반복문 마스터하기
  4. 함수와 람다 완벽 가이드
  5. 리스트, 튜플, 딕셔너리 정복하기
  6. 문자열 처리와 정규표현식
  7. 파일 입출력과 예외 처리

🚀 중급편 (8-12)

  1. 클래스와 객체지향 프로그래밍
  2. 모듈과 패키지 관리
  3. 데코레이터와 제너레이터 (현재 글)
  4. 비동기 프로그래밍 (async/await)
  5. 데이터베이스 연동하기

💼 고급편 (13-16)

  1. 웹 스크래핑과 API 개발
  2. 테스트와 디버깅 전략
  3. 성능 최적화 기법
  4. 멀티프로세싱과 병렬 처리

이전글: 모듈과 패키지 관리 ⬅️ 현재글: 데코레이터와 제너레이터 다음글: 비동기 프로그래밍 (async/await) ➡️

이번 포스트에서는 Python의 고급 기능인 데코레이터와 제너레이터를 완벽히 마스터했습니다. 다음 포스트에서는 비동기 프로그래밍에 대해 자세히 알아보겠습니다. Happy Coding! 🐍✨

Programming, Python
python programming tutorial beginner series decorators generators yield
이 기사는 저작권자의 CC BY 4.0 라이센스를 따릅니다.