[Python] 파이썬 Requests 모듈로 API 데이터 가져오기

요즘 대부분의 프로젝트에서 외부 API를 연동하는 일이 빈번하게 발생하는데, 처음에는 간단해 보이던 API 작업이 실무에선 여러 예외 상황과 에러 처리로 인해 생각보다 복잡해지곤 합니다. 오늘은 Requests 모듈을 효율적으로 사용하는 방법을 공유하려고 합니다.

목차

Requests 모듈 기본 개념과 설치 HTTP 메서드 활용하기 (GET, POST, PUT, DELETE) 헤더와 파라미터 설정 방법 효과적인 에러 핸들링과 예외 처리 인증 및 보안 처리 방법 API 호출 최적화와 모범 사례

Requests 모듈 기본 개념과 설치

파이썬으로 API 요청을 보내는 방법은 여러 가지가 있습니다. 표준 라이브러리의 urllib도 있고, httplib2나 HTTPX 같은 대안도 있죠. 그런데 왜 많은 개발자들이 Requests를 선호할까요? 단순함과 직관성 때문입니다. 정신 나간 복잡한 코드 없이 HTTP 요청을 가장 파이썬스럽게 처리할 수 있거든요.

Requests 모듈은 "인간을 위한 HTTP" 라는 슬로건을 내세울 정도로 사용성에 중점을 두고 있습니다. 복잡한 네트워크 통신 과정을 추상화해서 간단한 메서드 호출로 변환해주죠. 설치부터 시작해볼까요?

Requests 설치하기

pip를 이용해 간단하게 설치할 수 있습니다.

pip install requests

설치가 끝났다면 가장 기본적인 사용법부터 알아보겠습니다. 아래는 GET 요청을 보내는 가장 간단한 형태입니다.

import requests

response = requests.get('https://api.example.com/data')
print(response.status_code)  # 상태 코드 출력
print(response.text)  # 응답 본문 출력

이렇게 단 3줄로 API 요청을 보내고 응답을 확인할 수 있습니다. 별거 아닌 것 같지만, 이게 Requests의 철학이죠. 복잡한 일은 라이브러리가 처리하고, 개발자는 비즈니스 로직에 집중할 수 있게 해줍니다.

HTTP 메서드 활용하기 (GET, POST, PUT, DELETE)

RESTful API와 작업한다면 GET, POST, PUT, DELETE 같은 다양한 HTTP 메서드를 사용해야 할 겁니다. Requests 모듈은 각 메서드에 대응하는 함수를 제공해 이를 간단하게 처리할 수 있게 해줍니다.

HTTP 메서드	Requests 함수	일반적인 용도	예제 코드
GET	requests.get()	데이터 조회	requests.get('https://api.example.com/users')
POST	requests.post()	새 데이터 생성	requests.post('https://api.example.com/users', json=data)
PUT	requests.put()	데이터 수정(전체)	requests.put('https://api.example.com/users/1', json=data)
PATCH	requests.patch()	데이터 수정(일부)	requests.patch('https://api.example.com/users/1', json=data)
DELETE	requests.delete()	데이터 삭제	requests.delete('https://api.example.com/users/1')

각 메서드의 사용법을 자세히 알아보겠습니다.

GET 메서드 활용하기

GET은 데이터를 조회할 때 사용하는 가장 기본적인 메서드입니다. 쿼리 파라미터를 전달할 때는 params 인자를 사용하면 URL 인코딩을 자동으로 처리해줍니다.

import requests

# 쿼리 파라미터 설정
params = {
    'category': 'technology',
    'limit': 10,
    'sort': 'newest'
}

response = requests.get('https://api.example.com/articles', params=params)
print(response.url)  # 실제 요청 URL 확인
print(response.json())  # JSON 응답을 파이썬 객체로 변환

헤더와 파라미터 설정 방법

API 요청시 헤더 설정은 정말 중요합니다. 인증 토큰 전달부터 콘텐츠 타입 지정까지 다양한 용도로 사용되죠. Requests에서는 headers 인자를 통해 쉽게 설정할 수 있습니다.

자주 사용하는 헤더 설정

1. Content-Type: 요청 본문의 데이터 형식을 지정 (application/json, application/x-www-form-urlencoded 등)
2. Authorization: API 인증을 위한 토큰이나 자격 증명 전달
3. Accept: 서버에서 받고 싶은 응답 형식 지정 (application/json, text/html 등)
4. User-Agent: 클라이언트 정보 제공 (일부 API는 이 헤더를 체크함)
5. Cache-Control: 캐싱 동작 제어

다음은 헤더를 사용하는 실제 예제입니다:

import requests

headers = {
    'Authorization': 'Bearer your_access_token_here',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'User-Agent': 'MyApp/1.0.0'
}

response = requests.post(
    'https://api.example.com/data',
    headers=headers,
    json={'name': 'John Doe', 'email': 'john@example.com'}
)

# 응답 헤더 확인
print(response.headers)
print(response.headers.get('Content-Type'))

📝 데이터 전송 방식의 차이점

Requests에서 데이터를 전송할 때 json 파라미터와 data 파라미터의 차이를 아는 것이 중요합니다. json은 파이썬 객체를 JSON 문자열로 직렬화하고 Content-Type을 'application/json'으로 설정합니다. 반면 data는 일반적으로 폼 데이터로 전송되며 Content-Type을 'application/x-www-form-urlencoded'로 설정합니다. API 문서에서 요구하는 형식에 맞게 사용해야 합니다.

효과적인 에러 핸들링과 예외 처리

API 요청은 항상 성공하는 것이 아닙니다. 네트워크 문제, 서버 오류, 인증 실패 등 다양한 이유로 실패할 수 있습니다. 실무에서는 이런 예외 상황을 적절히 처리하는 것이 무엇보다 중요합니다. 많은 개발자들이 이 부분을 소홀히 하다가 프로덕션 환경에서 문제를 겪게 되죠.

HTTP 상태 코드 확인

가장 기본적인 에러 처리 방법은 응답의 상태 코드를 확인하는 것입니다. 주요 HTTP 상태 코드는 다음과 같습니다.

상태 코드	의미	처리 방법
2xx (200-299)	성공	응답 데이터 정상 처리
3xx (300-399)	리다이렉션	새 위치로 요청 재시도 (Requests는 자동 처리)
4xx (400-499)	클라이언트 오류	요청 내용 수정 필요 (인증, 파라미터 등)
5xx (500-599)	서버 오류	일정 시간 후 재시도 또는 서버 관리자에게 문의

Requests의 주요 예외 타입

더 세밀한 에러 처리를 위해 Requests가 발생시키는 다양한 예외 타입을 알아두면 좋습니다.

• ConnectionError: 네트워크 연결 실패 (DNS 실패, 거부된 연결 등)
• Timeout: 요청 시간 초과
• TooManyRedirects: 리다이렉션이 너무 많음
• RequestException: 기본 예외 클래스 (모든 Requests 예외의 부모 클래스)

인증 및 보안 처리 방법

대부분의 API는 어떤 형태로든 인증이 필요합니다. 인증 없이 누구나 접근할 수 있는 API는 보안적으로 취약하거나 데이터 남용의 위험이 있기 때문이죠. Requests는 다양한 인증 방식을 지원합니다.

일반적인 인증 방법들

인증 방식	설명	Requests 구현 방법
Basic Auth	사용자명과 비밀번호를 Base64로 인코딩	auth=('username', 'password')
Bearer Token	API 키나 액세스 토큰 사용	headers={'Authorization': 'Bearer token'}
API Key	쿼리 파라미터나 헤더로 API 키 전달	params={'api_key': 'your_key'}
OAuth	복잡한 토큰 기반 인증 프로토콜	일반적으로 별도 라이브러리 사용 (requests-oauthlib)

API 호출 최적화와 모범 사례

실무에서 API를 효율적으로 사용하려면 단순히 요청을 보내고 응답을 받는 것 이상의 최적화가 필요합니다. 특히 대규모 서비스에서는 API 호출 최적화가 성능과 비용에 직접적인 영향을 미칩니다.

• 타임아웃 설정: 요청이 무한정 대기하는 것을 방지하려면 항상 타임아웃을 설정하세요.
• 재시도 로직 구현: 일시적인 실패에 대응하기 위해 재시도 로직을 구현하세요.
• 캐싱 활용: 동일한 데이터를 반복해서 요청하지 않도록 캐싱을 구현하세요.
• 속도 제한 준수: API의 속도 제한(rate limit)을 준수하고 요청을 적절히 조절하세요.
• 연결 풀링: Session 객체를 사용하여 연결을 재사용하세요.

다음은 재시도 로직을 구현한 예제입니다:

import requests
import time
from requests.exceptions import RequestException

def request_with_retry(url, max_retries=3, backoff_factor=0.5, timeout=5):
    """
    재시도 로직이 포함된 요청 함수
    
    Args:
        url: 요청할 URL
        max_retries: 최대 재시도 횟수
        backoff_factor: 재시도 간격 증가 계수
        timeout: 요청 타임아웃 시간(초)
    
    Returns:
        응답 객체 또는 None(모든 시도 실패시)
    """
    retry_count = 0
    while retry_count < max_retries:
        try:
            response = requests.get(url, timeout=timeout)
            response.raise_for_status()
            return response
        except RequestException as e:
            retry_count += 1
            if retry_count >= max_retries:
                print(f"최대 재시도 횟수({max_retries})를 초과했습니다: {e}")
                return None
                
            # 지수 백오프: 재시도 간격을 점점 늘림
            wait_time = backoff_factor * (2 ** (retry_count - 1))
            print(f"재시도 {retry_count}/{max_retries}, {wait_time:.2f}초 후 다시 시도합니다: {e}")
            time.sleep(wait_time)
    
    return None

📝 API 클라이언트 클래스 만들기

대규모 프로젝트에서는 API 호출 로직을 클래스로 캡슐화하는 것이 좋습니다. 재사용성이 높아지고 코드가 더 깔끔해집니다. 예시로, 다음과 같이 API 클라이언트 클래스를 만들 수 있습니다:

class APIClient:
    def __init__(self, base_url, api_key=None, timeout=5):
        self.base_url = base_url
        self.session = requests.Session()
        if api_key:
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})
        self.timeout = timeout
        
    def get(self, endpoint, **kwargs):
        return self._request('GET', endpoint, **kwargs)
        
    def post(self, endpoint, data=None, json=None, **kwargs):
        return self._request('POST', endpoint, data=data, json=json, **kwargs)
        
    def _request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint}"
        kwargs.setdefault('timeout', self.timeout)
        return self.session.request(method, url, **kwargs)

📝 응답 데이터 다루기

JSON API 응답을 처리할 때 response.json() 메서드를 사용하면 JSON 문자열을 파이썬 객체로 자동 변환해줍니다. 실무에서는 API 응답 구조를 미리 확인하고, 원하는 데이터가 없을 경우를 대비해 get() 메서드나 예외 처리를 사용하세요.

# 안전한 데이터 추출
data = response.json()
user_name = data.get('user', {}).get('name', 'Unknown')  # 중첩 딕셔너리에서 안전하게 값 추출
items = data.get('items', [])  # 리스트가 없으면 빈 리스트 반환

실전에서 자주 사용하는 코드 예제

지금까지 배운 내용을 종합하여 실무에서 바로 활용할 수 있는 코드 예제를 살펴보겠습니다. 다음 예제는 외부 REST API에서 데이터를 가져오고, 처리하고, 결과를 반환하는 종합적인 예제입니다:

import requests
import time
import json
import logging
from requests.exceptions import RequestException

# 로깅 설정
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class WeatherAPIClient:
    """날씨 API 클라이언트 클래스"""
    
    def __init__(self, api_key, base_url="https://api.weatherapi.com/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        # 모든 요청에 공통으로 사용할 파라미터 설정
        self.session.params = {"key": self.api_key}
        
    def get_current_weather(self, location, retry=3, timeout=5):
        """현재 날씨 정보 조회
        
        Args:
            location: 위치(도시명, 우편번호, IP 주소 등)
            retry: 최대 재시도 횟수
            timeout: 요청 타임아웃(초)
            
        Returns:
            dict: 날씨 정보 또는 None(실패시)
        """
        endpoint = "current.json"
        params = {"q": location}
        
        for attempt in range(retry):
            try:
                logger.info(f"날씨 정보 요청 중... 위치: {location}, 시도: {attempt+1}/{retry}")
                response = self.session.get(
                    f"{self.base_url}/{endpoint}",
                    params=params,
                    timeout=timeout
                )
                response.raise_for_status()
                
                data = response.json()
                logger.info(f"날씨 정보 수신 성공: {location}")
                return {
                    "location": data["location"]["name"],
                    "country": data["location"]["country"],
                    "temperature_c": data["current"]["temp_c"],
                    "temperature_f": data["current"]["temp_f"],
                    "condition": data["current"]["condition"]["text"],
                    "humidity": data["current"]["humidity"],
                    "wind_kph": data["current"]["wind_kph"],
                    "last_updated": data["current"]["last_updated"]
                }
                
            except RequestException as e:
                logger.error(f"요청 실패 ({attempt+1}/{retry}): {str(e)}")
                if attempt == retry - 1:  # 마지막 시도였을 경우
                    logger.error(f"최대 재시도 횟수 초과. 위치: {location}")
                    return None
                # 지수 백오프 적용
                time.sleep(2 ** attempt)
                
            except (KeyError, ValueError) as e:
                logger.error(f"응답 데이터 파싱 실패: {str(e)}")
                return None
    
    def get_forecast(self, location, days=3):
        """날씨 예보 조회
        
        Args:
            location: 위치(도시명, 우편번호, IP 주소 등)
            days: 예보 일수 (1-10)
            
        Returns:
            dict: 예보 정보 또는 None(실패시)
        """
        endpoint = "forecast.json"
        params = {
            "q": location,
            "days": days,
            "aqi": "no",  # 대기질 정보 제외
            "alerts": "no"  # 기상 경보 제외
        }
        
        try:
            response = self.session.get(
                f"{self.base_url}/{endpoint}",
                params=params,
                timeout=10
            )
            response.raise_for_status()
            
            return response.json()
            
        except RequestException as e:
            logger.error(f"예보 데이터 요청 실패: {str(e)}")
            return None
    
    def close(self):
        """세션 종료"""
        self.session.close()
        
# 사용 예제
if __name__ == "__main__":
    # API 키는 환경 변수나 설정 파일에서 가져오는 것이 안전합니다
    API_KEY = "your_api_key_here" 
    
    # 클라이언트 인스턴스 생성
    client = WeatherAPIClient(API_KEY)
    
    try:
        # 현재 날씨 조회
        weather = client.get_current_weather("Seoul")
        if weather:
            print(f"{weather['location']}의 현재 날씨:")
            print(f"온도: {weather['temperature_c']}°C ({weather['temperature_f']}°F)")
            print(f"상태: {weather['condition']}")
            print(f"습도: {weather['humidity']}%")
            print(f"풍속: {weather['wind_kph']} km/h")
            print(f"마지막 업데이트: {weather['last_updated']}")
        
        # 예보 조회
        forecast = client.get_forecast("Tokyo", days=5)
        if forecast:
            print(f"\n{forecast['location']['name']}의 5일 예보:")
            for day in forecast['forecast']['forecastday']:
                date = day['date']
                max_temp = day['day']['maxtemp_c']
                min_temp = day['day']['mintemp_c']
                condition = day['day']['condition']['text']
                print(f"{date}: {min_temp}°C ~ {max_temp}°C, {condition}")
                
    except Exception as e:
        logger.error(f"예상치 못한 오류 발생: {str(e)}")
        
    finally:
        # 세션 정리
        client.close()

위 예제에서 볼 수 있듯이, 실제 프로젝트에서는 다음과 같은 패턴을 사용하는 것이 좋습니다.

1. API 클라이언트를 클래스로 캡슐화
2. 세션 재사용을 통한 성능 최적화
3. 체계적인 예외 처리 및 로깅
4. 재시도 메커니즘 구현
5. 응답 데이터 정제 및 변환

비동기 API 요청 처리하기

대량의 API 요청을 처리해야 하는 경우, 비동기 방식을 활용하면 성능을 크게 향상시킬 수 있습니다. 파이썬 3.5 이상에서는 asyncio와 aiohttp 라이브러리를 사용해 비동기 HTTP 요청을 구현할 수 있습니다:

import asyncio
import aiohttp
import time

async def fetch_data(session, url):
    """단일 URL에서 데이터 비동기적으로 가져오기"""
    async with session.get(url) as response:
        return await response.json()

async def fetch_all(urls):
    """여러 URL에서 동시에 데이터 가져오기"""
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, url) for url in urls]
        # 모든 요청을 동시에 실행하고 결과를 기다림
        return await asyncio.gather(*tasks)

# 사용 예제
if __name__ == "__main__":
    # 데이터를 가져올 URL 목록
    api_urls = [
        "https://api.example.com/users/1",
        "https://api.example.com/users/2",
        "https://api.example.com/users/3",
        "https://api.example.com/users/4",
        "https://api.example.com/users/5"
    ]
    
    # 순차 처리 시간 측정
    start_time = time.time()
    
    # requests로 순차 처리 (비교용)
    sequential_results = []
    for url in api_urls:
        response = requests.get(url)
        sequential_results.append(response.json())
    
    print(f"순차 처리 시간: {time.time() - start_time:.2f}초")
    
    # 비동기 처리 시간 측정
    start_time = time.time()
    
    # asyncio를 사용한 비동기 처리
    loop = asyncio.get_event_loop()
    parallel_results = loop.run_until_complete(fetch_all(api_urls))
    
    print(f"비동기 처리 시간: {time.time() - start_time:.2f}초")
    print(f"성능 향상: {len(api_urls)}개 요청 동시 처리")

비동기 요청은 I/O 작업이 많은 경우 특히 효과적입니다. API 호출은 대부분 네트워크 대기 시간이 길기 때문에 비동기 방식으로 구현하면 전체 실행 시간을 크게 단축할 수 있습니다.

마무리: Requests 모듈의 핵심 정리

파이썬 Requests 모듈은 API 통신의 복잡성을 추상화하여 직관적인 인터페이스를 제공합니다. 이 글에서 다룬 내용을 정리하자면:

• 기본적인 HTTP 메서드(GET, POST, PUT, DELETE) 사용 방법
• 헤더와 파라미터 설정을 통한 API 커스터마이징
• 다양한 인증 방식 구현 (Basic, Bearer 토큰, API 키)
• 체계적인 예외 처리와 에러 핸들링
• 세션을 활용한 최적화 및 연결 재사용
• 재시도 로직과 비동기 요청 처리를 통한 성능 향상

이제 여러분은 파이썬 Requests 모듈을 활용하여 다양한 API와 통신하고, 예외 상황을 효과적으로 처리하며, 대규모 시스템에서도 안정적으로 동작하는 코드를 작성할 수 있을 것입니다. 단순히 기능을 구현하는 것을 넘어, 효율적이고 안정적인 API 통신 코드를 작성하는 것이 실제 프로덕션 환경에서는 더 중요합니다. 감사합니다.