Python requests와 HTTP Streaming 문제 해결

HTTP Streaming을 제공하는 서버에 Python의 requests 라이브러리를 사용하여 GET 요청을 보낼 경우, 요청이 끝나지 않고 대기 상태가 지속되는 문제가 발생할 수 있다. 본 글에서는 이러한 문제의 원인과 해결 방법을 정리하고, 최적의 해결책을 제시한다.

---

1. 문제 상황

HTTP Streaming 방식으로 데이터를 지속적으로 전송하는 서버에 대해 requests.get()을 호출할 경우, 요청이 종료되지 않고 계속 응답을 받는 상태가 유지된다.

 

이로 인해 크롤러나 자동화된 스크립트가 다음 명령으로 진행하지 못하고 멈춰버리는 문제가 발생 가능

 

참고로 문제가 발생했던 대상의 경우 다음의 apache example 을 사용하고 있었음

https://github.com/jesperfj/buildpack-tomcat/blob/master/tomcat/webapps/examples/WEB-INF/classes/async/Stockticker.java

buildpack-tomcat/tomcat/webapps/examples/WEB-INF/classes/async/Stockticker.java at master · jesperfj/buildpack-tomcat

 

---

2. 원인 분석

이 문제는 다음과 같은 원인에 의해 발생한다.

1. 대상 서버가 HTTP Streaming 방식으로 응답
   • 서버가 한 번의 응답으로 끝내지 않고 지속적으로 데이터를 전송하기 때문에 요청이 종료되지 않는다.
2. Python requests의 timeout 설정 방식
   • requests.get(url, timeout=5)에서 timeout은 최초 연결 시간에만 적용되며, 이후 응답을 받는 시간에는 적용되지 않는다.
   • 따라서 서버가 응답을 지속적으로 보낸다면 timeout이 걸리지 않고 무한히 대기하게 된다.
3. requests의 기본 동작 방식
   • 특별한 설정 없이 requests.get()을 실행하면, 응답이 완전히 끝날 때까지 데이터를 기다리는 상태가 된다.
   • Streaming 방식에서는 응답이 끝나지 않으므로 요청이 계속 유지된다.
4. 자동화된 크롤러 또는 스크립트가 중단됨
   • 크롤러나 데이터 수집 코드가 특정 요청에서 멈춰버리고, 이후 작업을 수행하지 못하는 문제가 발생할 수 있다.

---

3. 시도해본 방법들

이 문제를 해결하기 위해 시도해볼 수 있는 세 가지 방법을 정리하고, 각각의 장단점을 분석한다.

1) signal.SIGALRM을 사용한 hard timeout

원리:

• signal.alarm()을 이용해 일정 시간이 지나면 SIGALRM을 발생시켜 요청을 강제로 종료한다.

코드 예시:

import requests
import signal

class TimeoutException(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutException("Request timed out")

signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(5)  # 5초 후 강제 타임아웃

try:
    response = requests.get("http://example.com/stockticker", stream=True)
    for line in response.iter_lines():
        print(line.decode("utf-8"))
except TimeoutException:
    print("Timeout occurred, terminating request.")
except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")
finally:
    signal.alarm(0)  # 타임아웃 해제

 

장점:

• requests.get()을 실행한 후 5초 후에 강제 종료 가능
• 코드가 간결하고 설정이 단순함

단점:

• SIGALRM은 Linux 및 Mac에서만 동작하며, Windows에서는 지원되지 않음
• 특정 요청마다 signal.alarm()을 조정 필요

평가:

• Linux 및 Mac 환경에서 실행한다면 가장 강력한 해결책
  • 해당 코드는 windows 에서 test 가 가능해야 했으므로 채택하지 않음
• 너무 tricky 하긴 하다.

---

2) thread.join()을 사용한 hard timeout

원리:

• 요청을 별도의 스레드에서 실행하고, thread.join(timeout=5)을 통해 일정 시간이 지나면 스레드를 종료하도록 한다.

코드 예시:

import requests
import threading

def fetch_data():
    try:
        response = requests.get("http://example.com/stockticker", stream=True)
        for line in response.iter_lines():
            print(line.decode("utf-8"))
    except requests.exceptions.RequestException as e:
        print(f"Request failed: {e}")

thread = threading.Thread(target=fetch_data)
thread.start()
thread.join(timeout=5)  # 5초 후 대기 종료

if thread.is_alive():
    print("Timeout reached, but request is still running.")

 

장점:

• Windows, Linux, Mac 모두에서 동작 가능
• 메인 프로세스를 블로킹하지 않고 요청을 실행 가능

단점:

• Python에서 스레드를 강제 종료할 방법이 없음
• 요청이 완료되지 않아 소켓이 계속 열려 있을 수 있음
  • lsof 로 확인해보면 TCP socket 을 계속 점유한다
• requests.get()이 내부적으로 실행되는 한, 해당 요청을 강제 종료할 방법이 없음

평가:

• 멀티플랫폼을 지원해야 한다면 고려할 수 있지만, 강제 종료가 불가능하여 비효율적

---

3) stream=True를 사용한 명시적 timeout 구현 -> 가장 추천

원리:

• stream=True를 사용하여 데이터를 한 번에 가져오지 않음
• 일정 시간(혹은 횟수) 지나면 response.close()를 호출하여 연결을 닫는다.

코드 예시:

import requests
import time

url = "http://example.com/stockticker"

try:
    with requests.get(url, stream=True, timeout=5) as response:
        start_time = time.time()
        
        for line in response.iter_lines():
            if time.time() - start_time > 5:
                print("Timeout reached, closing connection.")
                response.close()
                break  # 강제 종료
            
            if line:
                print(line.decode("utf-8"))

except requests.exceptions.RequestException as e:
    print(f"Request failed: {e}")

 

장점:

• Windows, Linux, Mac 모두에서 동작 가능
• requests의 기본 기능을 활용하므로 소켓을 안전하게 닫을 수 있음
• 불필요한 소켓 점유 문제 방지
• chucnk_size 를 조절해서 가져오는 응답 크기까지 제어 가능

단점:

• iter_lines()를 사용해야 하므로 코드가 다소 길어짐
• 요청이 진행되는 동안 타이머를 직접 체크해야 함 (=오버헤드)

평가:

• 멀티플랫폼을 지원하면서도 안전하게 요청을 종료할 수 있는 가장 현실적인 방법

---

4. 최종 결론

stream=True 를 쓰자

 

5. Wrapper 코드

 

방금까지 정리한 기능들은 회사에서 requests 기반 모듈들을 사용할 때 겪었던 내용이다.

 

즉, 저렇게 해결한 부분을 모든 requests 모듈 사용 부분에 대체해야 했다.

 

따라서, 다음과 같은 Wrapper 를 작성해서 일반적인 request 사용과 거의 흡사하게 만들어봤다.

 

아래의 wrapper 코드를 바로 import 해서 사용하면 일반적인 requests 와 비슷하게 사용하면서 hard timeout 가능

 

# default
import time

# typing
from typing import Optional

# pip
import requests as req


class Requests:
    def __init__(self, max_chunks: int = 10, chunk_timeout=10):
        self.max_chunks = max_chunks
        self.chunk_timeout = chunk_timeout

    def request(self, method: str, url: str, **kwargs) -> Optional[req.Response]:
        kwargs["stream"] = True  # Force stream mode
        response = req.request(method, url, **kwargs)
        content = bytearray()
        total_chunks = 0
        connect_time = time.time()

        try:
            # 1MB 씩
            for chunk in response.iter_content(chunk_size=1024 * 1024):
                # print(f"cnt -> {total_chunks}")
                if not chunk:
                    break

                if time.time() - connect_time > self.chunk_timeout:
                    # print("Chunk timeout reached. Closing connection.")
                    raise req.exceptions.Timeout("Max Timeout exceeded")

                content.extend(chunk)
                total_chunks += 1

                if total_chunks >= self.max_chunks:
                    # print("Stream limit reached. Closing connection.")
                    raise req.exceptions.Timeout("Hard Timeout (Chunk Count Exceeded)")

            response._content = bytes(content)
            return response

        except req.exceptions.Timeout as e:
            print(f"Timeout occurred: {e}")
            return None

        finally:
            response.close()
            # print("Connection closed.")

    def get(self, url: str, **kwargs) -> Optional[req.Response]:
        return self.request("GET", url, **kwargs)

    def post(self, url: str, data=None, json=None, **kwargs) -> Optional[req.Response]:
        return self.request("POST", url, data=data, json=json, **kwargs)

    def put(self, url: str, data=None, json=None, **kwargs) -> Optional[req.Response]:
        return self.request("PUT", url, data=data, json=json, **kwargs)

    def patch(self, url: str, data=None, json=None, **kwargs) -> Optional[req.Response]:
        return self.request("PATCH", url, data=data, json=json, **kwargs)

    def delete(
        self, url: str, data=None, json=None, **kwargs
    ) -> Optional[req.Response]:
        return self.request("DELETE", url, data=data, json=json, **kwargs)


# For Export
requests = Requests()