제너레이터(generator)

제너레이터란?:

• 이터레이터(Iterator)를 쉽게 작성할 수 있게 해주는 파이썬의 특별한 함수 혹은 표현식임.
• 일반적인 함수(def로 정의된)는 호출되면 내부의 로직을 모두 실행한 후 종료되지만, 제너레이터 함수는 함수 내에서 yield 키워드를 사용하여 실행 상태를 중간에 일시 정지(pause)하고, 호출자에게 값을 하나씩 반환한 뒤, 다시 해당 지점부터 실행을 재개할 수 있습니다.
• 즉, 제너레이터는 값을 한 번에 하나씩 생성하며, 전체 데이터를 미리 메모리에 로드하지 않아도 되므로,
  • 메모리 효율적
  • 데이터 스트리밍, 큰 데이터 처리, 또는 무한 시퀀스를 다루는 상황에서 큰 장점을 제공합니다.

 

제너레이터의 장점:

• 메모리 효율성
  • 모든 데이터를 한꺼번에 담지 않고, 필요한 순간마다 데이터를 하나씩 생성합니다.
  • 예: 100만 개의 값을 리스트로 만들 경우, 메모리에 100만 개 모두 올려야 하지만 제너레이터는 그때그때 1개씩만 반환하므로 훨씬 효율적입니다.
• 지연 평가(Lazy Evaluation)
  • 실제로 값이 필요한 시점에만 값을 생성합니다.
  • 이로 인해 계산 비용이 분산되어 초기 로드 시간이 단축될 수 있습니다.

 

제너레이터 함수 구현 방법:

def my_generator():
    print("첫 번째 실행 부분")
    yield 1  # 첫 번째 값 반환 후, 함수가 여기서 잠시 중단된다.

    print("두 번째 실행 부분")
    yield 2  # 두 번째 값 반환 후, 다시 잠시 중단.

    print("세 번째 실행 부분")
    yield 3  # 세 번째 값 반환 후, 종료.

gen = my_generator()  # 제너레이터 객체 생성

print(next(gen))  # "첫 번째 실행 부분" 출력 후, 1 반환
print(next(gen))  # "두 번째 실행 부분" 출력 후, 2 반환
print(next(gen))  # "세 번째 실행 부분" 출력 후, 3 반환
# 그 다음 next(gen)을 호출하면 StopIteration 예외가 발생

• 함수 안에서 yield가 실행될 때마다 해당 값을 반환하고, 실행 상태가 바로 다음 줄에서 일시 정지됩니다.
• 이후에 또 값을 요청(next())하면, 직전에 멈췄던 지점에서 실행이 재개됩니다.

 

yield와 함수 종료:

def countdown(n):
    while n > 0:
        yield n
        n -= 1
    return  # 혹은 명시적으로 return x 처럼 값 반환 시도 가능(파이썬 3.3+), 
            # 하지만 호출 측에서는 StopIteration 예외로 처리됨

• return을 사용하면 그 즉시 함수가 종료되므로, 제너레이터도 그 시점에서 종료됩니다.
• 제너레이터 안에서 yield 키워드는 여러 번 등장할 수 있습니다.

 

제너레이터 표현식(Generator Expression):

# 예: 0부터 999,999까지의 제곱을 생성하는 제너레이터
squares = (x*x for x in range(1000000))

• 리스트 컴프리헨션과 비슷하지만, 소괄호(()) 를 사용하며, 즉석에서 제너레이터를 만들어내는 문법입니다.
• 위 코드에서 squares는 리스트가 아닌 제너레이터로, 실제로 값이 필요할 때마다 순차적으로 값을 생성합니다.
• 리스트 컴프리헨션 [x*x for x in range(1000000)]를 사용하면 모든 값을 한 번에 메모리에 올리게 되므로, 메모리 사용량이 훨씬 커집니다.

 

고급 사용법: yield from:

def sub_generator():
    yield 1
    yield 2
    yield 3

def main_generator():
    # sub_generator가 반환하는 모든 값을 하나씩 yield
    yield from sub_generator()  
    yield 4
    yield 5

for value in main_generator():
    print(value)

# 출력:
# 1
# 2
# 3
# 4
# 5

• 파이썬 3.3 이상에서는 yield from 키워드로 하위 제너레이터를 간편하게 위임할 수 있습니다.
• yield from은 다른 제너레이터 혹은 이터러블 객체를 반복하여 값들을 차례로 내보냅니다.
• 제너레이터들을 합성하거나 중첩된 구조를 단순화할 때 유용합니다.

 

제너레이터와 이터레이터, 이터러블:

• 이터러블(Iterable): __iter__() 메서드를 가지고 있어서, 반복(for문 등) 가능한 객체
• 이터레이터(Iterator): __next__() 메서드를 통해 다음 값을 반환하고, 더 이상 반환할 값이 없으면 StopIteration 예외를 발생시키는 객체
• 제너레이터(Generator): 파이썬에서 이터레이터를 간단히 구현하기 위한 특별한 함수 또는 표현식
  • 자동으로 __iter__()와 __next__()를 지원하므로, 이터레이션 가능한 객체를 직접 구현하는 번거로움을 줄여 줍니다.

 

제너레이터 사용 예시

큰 파일의 라인을 하나씩 처리하기:

def read_large_file(file_path):
    with open(file_path, 'r') as f:
        for line in f:
            yield line.strip()

for line_data in read_large_file("large_data.txt"):
    # 메모리 효율적으로 한 줄씩 읽어 처리
    print(line_data)

 

API 응답 스트리밍:

from flask import Response

def generate_data():
    for i in range(1, 1000001):
        yield f"data: {i}\n"

@app.route('/stream')
def stream():
    return Response(generate_data(), mimetype='text/plain')

• 대량의 데이터를 반환해야 하는 상황에서, 한꺼번에 모든 JSON 데이터를 생성하기보다는 필요한 부분만 끊어서 반환하는 식으로 사용할 수 있습니다.
• 플라스크(Flask), 파스트API(FastAPI) 등에서 제너레이터를 응답에 활용하면, 클라이언트는 데이터를 스트리밍 형태로 수신할 수 있습니다.
• 주의) 제너레이터를 사용해 응답을 보내면 HTTP 응답 스트리밍이 되지만, 기본적으로 SSE(Server-Sent Events) 방식이 되지는 않습니다:
  • Flask나 FastAPI에서 제너레이터를 이용해 응답을 스트리밍하는 경우, 이는 단순한 HTTP 응답 스트리밍입니다. 하지만 SSE(Server-Sent Events)는 HTTP 응답 스트리밍을 이용하는 특정한 프로토콜이기 때문에, 적절한 헤더 및 포맷을 맞춰야 합니다.
  • 제너레이터를 이용해 응답을 반환하면, HTTP 응답이 chunked transfer encoding 방식으로 클라이언트에 전달됩니다.

 

배경지식 - SSE 를 위한 조건:

• HTTP 헤더를 설정해야 함
  • Content-Type: text/event-stream
  • Cache-Control: no-cache
  • Connection: keep-alive
• 데이터를 특정 형식으로 전송해야 함
  • 각 메시지는 data: 메시지\n\n 형식으로 전송해야 함

 

Chunk 단위 버퍼링(합쳐서 전송) 패턴:

def buffered_stream_text(prompt, chunk_size=5):
    buffer = []
    for token in generate_tokens(prompt):
        buffer.append(token)
        if len(buffer) >= chunk_size:
            yield ''.join(buffer)
            buffer = []
    # 마지막에 남은 토큰 처리
    if buffer:
        yield ''.join(buffer)

 

pytest 내부 설정 관리 - 가상의 예시 wrapper:

class FixtureManager:
    def __init__(self):
        self.fixtures = {}
        
    def fixture(self, func):
        """fixture 데코레이터 구현"""
        def wrapper(*args, **kwargs):
            # 제너레이터 함수 실행
            gen = func(*args, **kwargs)
            
            try:
                # 1. Setup & 값 얻기
                fixture_value = next(gen)
                
                try:
                    # 2. 테스트에 값 전달
                    return fixture_value
                    
                finally:
                    try:
                        # 3. 정리(cleanup) 실행
                        next(gen)
                    except StopIteration:
                        pass
                        
            except StopIteration:
                return None