HTTP 요청

Python의 requests 라이브러리로 웹 API와 통신하는 방법을 배워봅시다.

requests란?

requests는 Python에서 HTTP 요청을 보내기 위한 가장 인기 있는 라이브러리입니다. 간단하고 직관적인 API로 웹 서버와 쉽게 통신할 수 있습니다.

주요 특징

• 간단하고 직관적인 API
• 자동 JSON 인코딩/디코딩
• 세션 관리 및 쿠키 처리
• 파일 업로드 지원
• 타임아웃 설정 가능

설치하기

pip install requests

기본 GET 요청

가장 기본적인 GET 요청부터 시작해봅시다.

import requests

# 기본 GET 요청
response = requests.get('https://api.github.com')

# 응답 확인
print(response.status_code)  # 200
print(response.text)         # 응답 본문
print(response.json())       # JSON 형식으로 파싱

쿼리 파라미터 추가

# URL에 직접 작성
response = requests.get('https://api.github.com/search/repositories?q=python')

# params 딕셔너리 사용 (권장)
params = {
    'q': 'python',
    'sort': 'stars',
    'order': 'desc'
}
response = requests.get('https://api.github.com/search/repositories', params=params)

print(response.url)  # 실제 요청된 URL 확인

POST 요청

데이터를 서버로 전송할 때 POST 요청을 사용합니다.

# Form 데이터 전송
data = {
    'username': 'john',
    'password': 'secret123'
}
response = requests.post('https://httpbin.org/post', data=data)

# JSON 데이터 전송
json_data = {
    'name': 'John Doe',
    'email': 'john@example.com',
    'age': 30
}
response = requests.post('https://httpbin.org/post', json=json_data)

print(response.json())

파일 업로드

# 단일 파일 업로드
files = {'file': open('report.pdf', 'rb')}
response = requests.post('https://httpbin.org/post', files=files)

# 파일명 지정
files = {
    'file': ('report.pdf', open('report.pdf', 'rb'), 'application/pdf')
}
response = requests.post('https://httpbin.org/post', files=files)

# 여러 파일 업로드
files = {
    'file1': open('image1.jpg', 'rb'),
    'file2': open('image2.jpg', 'rb')
}
response = requests.post('https://httpbin.org/post', files=files)

PUT과 DELETE 요청

리소스를 업데이트하거나 삭제할 때 사용합니다.

# PUT 요청 (전체 업데이트)
update_data = {
    'id': 1,
    'title': 'Updated Title',
    'completed': True
}
response = requests.put('https://jsonplaceholder.typicode.com/todos/1', json=update_data)

# PATCH 요청 (부분 업데이트)
partial_data = {'completed': True}
response = requests.patch('https://jsonplaceholder.typicode.com/todos/1', json=partial_data)

# DELETE 요청
response = requests.delete('https://jsonplaceholder.typicode.com/todos/1')
print(response.status_code)  # 200 또는 204

헤더 설정

요청 헤더를 통해 추가 정보를 전달할 수 있습니다.

# 기본 헤더 설정
headers = {
    'User-Agent': 'MyApp/1.0',
    'Accept': 'application/json',
    'Content-Type': 'application/json'
}
response = requests.get('https://api.github.com', headers=headers)

# API 키 인증
headers = {
    'Authorization': 'Bearer YOUR_API_KEY',
    'X-API-Key': 'YOUR_API_KEY'
}
response = requests.get('https://api.example.com/data', headers=headers)

인증

다양한 인증 방법을 지원합니다.

from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# Basic 인증
response = requests.get(
    'https://api.example.com/protected',
    auth=HTTPBasicAuth('username', 'password')
)

# 간단한 방법
response = requests.get(
    'https://api.example.com/protected',
    auth=('username', 'password')
)

# Bearer Token 인증
headers = {'Authorization': 'Bearer YOUR_ACCESS_TOKEN'}
response = requests.get('https://api.example.com/data', headers=headers)

# OAuth 2.0
from requests_oauthlib import OAuth2Session

oauth = OAuth2Session(client_id, token=token)
response = oauth.get('https://api.example.com/protected')

JSON 처리

JSON 데이터를 쉽게 다룰 수 있습니다.

# JSON 응답 파싱
response = requests.get('https://api.github.com/users/github')
data = response.json()

print(data['name'])
print(data['public_repos'])

# JSON 요청 전송
payload = {
    'name': 'John',
    'email': 'john@example.com',
    'settings': {
        'notifications': True,
        'theme': 'dark'
    }
}
response = requests.post('https://api.example.com/users', json=payload)

# 커스텀 JSON 인코더
import json

class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        return super().default(obj)

response = requests.post(
    'https://api.example.com/data',
    data=json.dumps(payload, cls=CustomEncoder),
    headers={'Content-Type': 'application/json'}
)

타임아웃 설정

네트워크 문제로 인한 무한 대기를 방지합니다.

# 연결 타임아웃 5초, 읽기 타임아웃 10초
try:
    response = requests.get('https://api.example.com', timeout=(5, 10))
except requests.exceptions.Timeout:
    print("요청 시간이 초과되었습니다.")

# 전체 타임아웃 10초
response = requests.get('https://api.example.com', timeout=10)

# 타임아웃 없음 (권장하지 않음)
response = requests.get('https://api.example.com', timeout=None)

에러 처리

다양한 HTTP 에러를 적절히 처리해야 합니다.

import requests
from requests.exceptions import (
    ConnectionError,
    Timeout,
    HTTPError,
    RequestException
)

try:
    response = requests.get('https://api.example.com/data', timeout=5)
    response.raise_for_status()  # 4xx, 5xx 에러 발생

    data = response.json()
    print(data)

except ConnectionError:
    print("연결 실패: 네트워크를 확인하세요.")
except Timeout:
    print("시간 초과: 서버가 응답하지 않습니다.")
except HTTPError as e:
    print(f"HTTP 에러: {e.response.status_code}")
    if e.response.status_code == 404:
        print("리소스를 찾을 수 없습니다.")
    elif e.response.status_code == 401:
        print("인증이 필요합니다.")
    elif e.response.status_code == 500:
        print("서버 내부 오류입니다.")
except RequestException as e:
    print(f"요청 실패: {str(e)}")

세션 사용

여러 요청을 보낼 때 세션을 사용하면 효율적입니다.

# 세션 생성
session = requests.Session()

# 모든 요청에 적용될 헤더 설정
session.headers.update({
    'User-Agent': 'MyApp/1.0',
    'Authorization': 'Bearer YOUR_TOKEN'
})

# 세션으로 여러 요청
response1 = session.get('https://api.example.com/users')
response2 = session.get('https://api.example.com/posts')
response3 = session.post('https://api.example.com/comments', json={'text': 'Hello'})

# 쿠키 자동 관리
login_data = {'username': 'john', 'password': 'secret'}
session.post('https://example.com/login', data=login_data)

# 로그인 후 인증된 요청
response = session.get('https://example.com/dashboard')

# 세션 종료
session.close()

# Context Manager 사용 (권장)
with requests.Session() as session:
    session.headers.update({'Authorization': 'Bearer TOKEN'})
    response = session.get('https://api.example.com/data')

재시도 설정

네트워크 문제 시 자동으로 재시도할 수 있습니다.

from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry

# 재시도 전략 설정
retry_strategy = Retry(
    total=3,                    # 최대 3번 재시도
    backoff_factor=1,           # 1초, 2초, 4초 대기
    status_forcelist=[429, 500, 502, 503, 504],
    method_whitelist=["HEAD", "GET", "OPTIONS", "POST"]
)

adapter = HTTPAdapter(max_retries=retry_strategy)
session = requests.Session()
session.mount("https://", adapter)
session.mount("http://", adapter)

# 자동 재시도 적용
response = session.get('https://api.example.com/data')

실전 예제

예제 1: OpenWeather API로 날씨 정보 가져오기

import requests

def get_weather(city, api_key):
    """특정 도시의 날씨 정보를 가져옵니다."""
    base_url = "http://api.openweathermap.org/data/2.5/weather"

    params = {
        'q': city,
        'appid': api_key,
        'units': 'metric',  # 섭씨 온도
        'lang': 'kr'        # 한국어
    }

    try:
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()

        data = response.json()

        weather_info = {
            'city': data['name'],
            'temperature': data['main']['temp'],
            'feels_like': data['main']['feels_like'],
            'humidity': data['main']['humidity'],
            'description': data['weather'][0]['description'],
            'wind_speed': data['wind']['speed']
        }

        return weather_info

    except requests.exceptions.RequestException as e:
        print(f"날씨 정보를 가져오는데 실패했습니다: {e}")
        return None

# 사용 예시
api_key = "YOUR_API_KEY"
weather = get_weather("Seoul", api_key)

if weather:
    print(f"도시: {weather['city']}")
    print(f"온도: {weather['temperature']}°C")
    print(f"체감온도: {weather['feels_like']}°C")
    print(f"습도: {weather['humidity']}%")
    print(f"날씨: {weather['description']}")
    print(f"풍속: {weather['wind_speed']} m/s")

예제 2: GitHub API로 저장소 정보 조회

import requests
from typing import List, Dict

class GitHubAPI:
    def __init__(self, token=None):
        self.base_url = "https://api.github.com"
        self.session = requests.Session()

        if token:
            self.session.headers.update({
                'Authorization': f'token {token}',
                'Accept': 'application/vnd.github.v3+json'
            })

    def get_user_info(self, username: str) -> Dict:
        """사용자 정보를 가져옵니다."""
        url = f"{self.base_url}/users/{username}"
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()

    def get_user_repos(self, username: str) -> List[Dict]:
        """사용자의 저장소 목록을 가져옵니다."""
        url = f"{self.base_url}/users/{username}/repos"
        params = {
            'sort': 'updated',
            'per_page': 10
        }
        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()

    def search_repositories(self, query: str, language=None) -> List[Dict]:
        """저장소를 검색합니다."""
        url = f"{self.base_url}/search/repositories"

        search_query = query
        if language:
            search_query += f" language:{language}"

        params = {
            'q': search_query,
            'sort': 'stars',
            'order': 'desc',
            'per_page': 10
        }

        response = self.session.get(url, params=params)
        response.raise_for_status()
        return response.json()['items']

# 사용 예시
github = GitHubAPI()

# 사용자 정보 조회
user = github.get_user_info('torvalds')
print(f"이름: {user['name']}")
print(f"팔로워: {user['followers']}")
print(f"공개 저장소: {user['public_repos']}")

# 저장소 검색
repos = github.search_repositories('machine learning', language='python')
for repo in repos[:5]:
    print(f"\n{repo['full_name']}")
    print(f"⭐ {repo['stargazers_count']} | 🍴 {repo['forks_count']}")
    print(f"설명: {repo['description']}")

예제 3: REST API 클라이언트 만들기

import requests
from typing import Optional, Dict, Any
import json

class APIClient:
    """범용 REST API 클라이언트"""

    def __init__(self, base_url: str, api_key: Optional[str] = None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()

        # 기본 헤더 설정
        self.session.headers.update({
            'Content-Type': 'application/json',
            'User-Agent': 'Python-API-Client/1.0'
        })

        # API 키가 있으면 헤더에 추가
        if api_key:
            self.session.headers.update({'Authorization': f'Bearer {api_key}'})

    def _make_request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> requests.Response:
        """HTTP 요청을 보냅니다."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"

        try:
            response = self.session.request(method, url, timeout=30, **kwargs)
            response.raise_for_status()
            return response
        except requests.exceptions.HTTPError as e:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            print(f"Request Error: {str(e)}")
            raise

    def get(self, endpoint: str, params: Optional[Dict] = None) -> Dict[str, Any]:
        """GET 요청"""
        response = self._make_request('GET', endpoint, params=params)
        return response.json()

    def post(self, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        """POST 요청"""
        response = self._make_request('POST', endpoint, json=data)
        return response.json()

    def put(self, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        """PUT 요청"""
        response = self._make_request('PUT', endpoint, json=data)
        return response.json()

    def patch(self, endpoint: str, data: Optional[Dict] = None) -> Dict[str, Any]:
        """PATCH 요청"""
        response = self._make_request('PATCH', endpoint, json=data)
        return response.json()

    def delete(self, endpoint: str) -> bool:
        """DELETE 요청"""
        response = self._make_request('DELETE', endpoint)
        return response.status_code in [200, 204]

# 사용 예시
client = APIClient('https://jsonplaceholder.typicode.com')

# GET 요청
posts = client.get('/posts', params={'userId': 1})
print(f"게시글 수: {len(posts)}")

# POST 요청
new_post = client.post('/posts', data={
    'title': 'New Post',
    'body': 'This is a new post',
    'userId': 1
})
print(f"생성된 게시글 ID: {new_post['id']}")

# PUT 요청
updated_post = client.put('/posts/1', data={
    'id': 1,
    'title': 'Updated Title',
    'body': 'Updated body',
    'userId': 1
})

# DELETE 요청
success = client.delete('/posts/1')
print(f"삭제 성공: {success}")

예제 4: 다운로드 진행률 표시

import requests
from tqdm import tqdm

def download_file(url: str, filename: str):
    """파일을 다운로드하고 진행률을 표시합니다."""
    response = requests.get(url, stream=True)
    response.raise_for_status()

    total_size = int(response.headers.get('content-length', 0))

    with open(filename, 'wb') as file, tqdm(
        desc=filename,
        total=total_size,
        unit='B',
        unit_scale=True,
        unit_divisor=1024,
    ) as progress_bar:
        for chunk in response.iter_content(chunk_size=8192):
            file.write(chunk)
            progress_bar.update(len(chunk))

    print(f"다운로드 완료: {filename}")

# 사용 예시
url = "https://example.com/large-file.zip"
download_file(url, "downloaded-file.zip")

자주 묻는 질문

Q1. requests와 urllib의 차이점은?

A: requests는 urllib 위에 구축된 더 사용하기 쉬운 라이브러리입니다. requests는 간단한 API, 자동 JSON 처리, 세션 관리 등의 편의 기능을 제공합니다.

# urllib (복잡함)
import urllib.request
import json

request = urllib.request.Request(
    'https://api.github.com',
    headers={'User-Agent': 'Python'}
)
response = urllib.request.urlopen(request)
data = json.loads(response.read().decode())

# requests (간단함)
import requests

response = requests.get('https://api.github.com')
data = response.json()

Q2. 대용량 파일은 어떻게 다운로드하나요?

A: stream=True 옵션을 사용하여 청크 단위로 다운로드합니다.

response = requests.get(url, stream=True)
with open('large_file.zip', 'wb') as f:
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            f.write(chunk)

Q3. SSL 인증서 오류가 발생합니다.

A: 개발 환경에서는 verify=False를 사용할 수 있지만, 프로덕션에서는 권장하지 않습니다.

# 개발 환경에서만 사용
response = requests.get(url, verify=False)

# 프로덕션: 인증서 경로 지정
response = requests.get(url, verify='/path/to/certfile')

Q4. 프록시는 어떻게 설정하나요?

A: proxies 매개변수를 사용합니다.

proxies = {
    'http': 'http://10.10.10.10:8000',
    'https': 'http://10.10.10.10:8000',
}

response = requests.get('https://api.example.com', proxies=proxies)

Q5. 응답이 JSON이 아닐 때는?

A: 응답 유형에 따라 적절한 속성을 사용합니다.

response = requests.get(url)

# JSON
data = response.json()

# 텍스트
text = response.text

# 바이너리
binary = response.content

# 자동 감지
if 'application/json' in response.headers.get('Content-Type', ''):
    data = response.json()
else:
    text = response.text

다음 단계

requests로 HTTP 요청을 마스터했다면, 다음 주제를 학습해보세요:

1. 웹 스크래핑 - BeautifulSoup으로 웹 페이지에서 데이터 추출하기
2. 비동기 요청 - aiohttp로 동시에 여러 요청 처리하기
3. API 개발 - FastAPI로 자신만의 REST API 만들기
4. 인증 - OAuth 2.0, JWT 등 고급 인증 방법

참고 자료

• Requests 공식 문서
• HTTP 상태 코드
• REST API 디자인 가이드