KR_HTTP - somaz94/python-study GitHub Wiki

Python HTTP 톡신 κ°œλ… 정리


1️⃣ HTTP 기초

HTTPλŠ” ν΄λΌμ΄μ–ΈνŠΈμ™€ μ„œλ²„ κ°„μ˜ 톡신을 μœ„ν•œ ν”„λ‘œν† μ½œμ΄λ‹€.

import requests
import json

# κΈ°λ³Έ GET μš”μ²­
response = requests.get('https://api.example.com/data')
print(f"μƒνƒœ μ½”λ“œ: {response.status_code}")  # μƒνƒœ μ½”λ“œ
print(f"응닡 헀더: {response.headers}")     # 응닡 헀더
print(f"응닡 λ‚΄μš©: {response.text}")        # 응닡 λ‚΄μš©
print(f"JSON 응닡: {response.json()}")      # JSON 응닡 νŒŒμ‹±

# 응닡 μƒνƒœ 확인
if response.status_code == 200:
    print("μš”μ²­ 성곡!")
elif response.status_code == 404:
    print("λ¦¬μ†ŒμŠ€λ₯Ό 찾을 수 μ—†μŠ΅λ‹ˆλ‹€")
else:
    print(f"였λ₯˜ λ°œμƒ: {response.status_code}")

# κΈ°λ³Έ POST μš”μ²­
data = {'username': 'user1', 'email': '[email protected]'}
response = requests.post('https://api.example.com/users', json=data)

# multipart/form-data μš”μ²­ (파일 μ—…λ‘œλ“œ)
files = {'file': open('document.pdf', 'rb')}
form_data = {'description': 'μ€‘μš” λ¬Έμ„œ'}
response = requests.post(
    'https://api.example.com/upload', 
    files=files, 
    data=form_data
)

# PUT μš”μ²­ (λ¦¬μ†ŒμŠ€ μˆ˜μ •)
update_data = {'username': 'updated_user', 'email': '[email protected]'}
response = requests.put('https://api.example.com/users/123', json=update_data)

# DELETE μš”μ²­ (λ¦¬μ†ŒμŠ€ μ‚­μ œ)
response = requests.delete('https://api.example.com/users/123')

# νƒ€μž„μ•„μ›ƒ μ„€μ •
try:
    response = requests.get('https://api.example.com/data', timeout=5)  # 5초 νƒ€μž„μ•„μ›ƒ
except requests.exceptions.Timeout:
    print("μš”μ²­ μ‹œκ°„μ΄ μ΄ˆκ³Όλ˜μ—ˆμŠ΅λ‹ˆλ‹€")

βœ… νŠΉμ§•:

  • κ°„λ‹¨ν•œ μš”μ²­/응닡
  • JSON 처리
  • μƒνƒœ μ½”λ“œ 확인
  • λ‹€μ–‘ν•œ HTTP λ©”μ„œλ“œ 지원
  • νƒ€μž„μ•„μ›ƒ μ„€μ •
  • 파일 μ—…λ‘œλ“œ
  • μƒνƒœ μ½”λ“œ 처리

HTTP λ©”μ„œλ“œ:

HTTP ν†΅μ‹ μ—μ„œ μ‚¬μš©λ˜λŠ” μ£Όμš” λ©”μ„œλ“œμ™€ μš©λ„μ΄λ‹€.

λ©”μ„œλ“œ λͺ©μ  νŠΉμ§• Python 예제
GET 데이터 쑰회 데이터λ₯Ό URL에 포함, 캐싱 κ°€λŠ₯ requests.get(url)
POST 데이터 생성 데이터λ₯Ό 본문에 포함, 캐싱 λΆˆκ°€ requests.post(url, json=data)
PUT 데이터 μˆ˜μ • λ¦¬μ†ŒμŠ€ 전체 λŒ€μ²΄, λ©±λ“±μ„± requests.put(url, json=data)
PATCH 데이터 λΆ€λΆ„ μˆ˜μ • λ¦¬μ†ŒμŠ€ 일뢀 μˆ˜μ • requests.patch(url, json=data)
DELETE 데이터 μ‚­μ œ λ¦¬μ†ŒμŠ€ 제거 requests.delete(url)
HEAD ν—€λ”λ§Œ 쑰회 λ³Έλ¬Έ 없이 ν—€λ”λ§Œ λ°˜ν™˜ requests.head(url)
OPTIONS 지원 μ˜΅μ…˜ 확인 μ„œλ²„ 지원 μ˜΅μ…˜ 확인 requests.options(url)


2️⃣ μš”μ²­ 헀더와 λ§€κ°œλ³€μˆ˜

HTTP μš”μ²­μ— 헀더와 λ§€κ°œλ³€μˆ˜λ₯Ό μΆ”κ°€ν•˜μ—¬ μ„œλ²„μ™€μ˜ 톡신을 μ»€μŠ€ν„°λ§ˆμ΄μ§•ν•˜λŠ” 방법이닀.

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# 헀더 μΆ”κ°€
headers = {
    'User-Agent': 'Python/3.9 Requests/2.26.0',
    'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...',
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    'X-Request-ID': '123e4567-e89b-12d3-a456-426614174000',
    'Cache-Control': 'no-cache'
}

# URL λ§€κ°œλ³€μˆ˜ (쿼리 νŒŒλΌλ―Έν„°)
params = {
    'page': 1,
    'limit': 10,
    'sort': 'created_at',
    'order': 'desc',
    'filter': 'active'
}

# κΈ°λ³Έ μš”μ²­
response = requests.get(
    'https://api.example.com/users',
    headers=headers,
    params=params
)

print(f"μ΅œμ’… URL: {response.url}")  # 쿼리 νŒŒλΌλ―Έν„°κ°€ ν¬ν•¨λœ URL 좜λ ₯

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

# λ‹€μ΄μ œμŠ€νŠΈ 인증 (Digest Authentication)
response = requests.get(
    'https://api.example.com/protected',
    auth=HTTPDigestAuth('username', 'password')
)

# API ν‚€ 인증
api_key = '1234567890abcdef'
# 쿼리 νŒŒλΌλ―Έν„°λ‘œ API ν‚€ 전달
response = requests.get(
    'https://api.example.com/data',
    params={'api_key': api_key}
)

# ν—€λ”λ‘œ API ν‚€ 전달
response = requests.get(
    'https://api.example.com/data',
    headers={'X-API-Key': api_key}
)

# OAuth 2.0 인증
token = 'access_token_here'
response = requests.get(
    'https://api.example.com/me',
    headers={'Authorization': f'Bearer {token}'}
)

# μΏ ν‚€ μ„€μ •
cookies = {'session_id': '12345', 'user_id': '42'}
response = requests.get(
    'https://api.example.com/dashboard',
    cookies=cookies
)

# 응닡 μΏ ν‚€ 확인
print(response.cookies)
for cookie in response.cookies:
    print(f"{cookie.name}: {cookie.value}")

βœ… νŠΉμ§•:

  • 헀더 μ»€μŠ€ν„°λ§ˆμ΄μ§•
  • λ§€κ°œλ³€μˆ˜ 전달
  • 인증 처리
  • μΏ ν‚€ 관리
  • API ν‚€ 인증
  • OAuth 지원
  • μ»€μŠ€ν…€ 헀더

일반적인 HTTP 헀더:

HTTP ν†΅μ‹ μ—μ„œ 자주 μ‚¬μš©λ˜λŠ” μš”μ²­ 및 응닡 헀더이닀.

헀더 μš©λ„ μ˜ˆμ‹œ
Content-Type 데이터 ν˜•μ‹ μ§€μ • application/json, multipart/form-data
Authorization 인증 정보 전달 Bearer token123, Basic base64encoded
User-Agent ν΄λΌμ΄μ–ΈνŠΈ 식별 Python/3.9 Requests/2.26.0
Accept 응닡 ν˜•μ‹ μš”μ²­ application/json, text/html
Cache-Control 캐싱 λ™μž‘ μ œμ–΄ no-cache, max-age=3600
Cookie μΏ ν‚€ 전솑 session_id=abc123
X-Request-ID μš”μ²­ 좔적 123e4567-e89b-12d3-a456-426614174000


3️⃣ μ„Έμ…˜ 관리

μ„Έμ…˜μ„ μ‚¬μš©ν•˜μ—¬ μ—¬λŸ¬ μš”μ²­ 간에 μƒνƒœλ₯Ό μœ μ§€ν•˜κ³  효율적으둜 HTTP 톡신을 κ΄€λ¦¬ν•˜λŠ” 방법이닀.

import requests
import time

# μ„Έμ…˜ 생성
session = requests.Session()

# μ„Έμ…˜μ— κΈ°λ³Έ 헀더 μ„€μ •
session.headers.update({
    'User-Agent': 'Python/3.9 SessionClient/1.0',
    'Authorization': 'Bearer token123',
    'Accept': 'application/json'
})

# μ„Έμ…˜μ— κΈ°λ³Έ μΏ ν‚€ μ„€μ •
session.cookies.update({'user_preference': 'dark_mode'})

# μ„Έμ…˜μ„ ν†΅ν•œ μš”μ²­ - λͺ¨λ“  μš”μ²­μ— 헀더와 μΏ ν‚€κ°€ μžλ™μœΌλ‘œ 포함됨
response = session.get('https://api.example.com/profile')
print(f"ν”„λ‘œν•„ 응닡: {response.status_code}")

# μ„Έμ…˜μ— μΆ”κ°€ λ§€κ°œλ³€μˆ˜ μ„€μ •
response = session.get(
    'https://api.example.com/articles',
    params={'category': 'technology', 'page': 1}
)
print(f"기사 응닡: {response.status_code}")

# μ„Έμ…˜μ„ ν†΅ν•œ POST μš”μ²­
response = session.post(
    'https://api.example.com/comments',
    json={'article_id': 42, 'text': '쒋은 κΈ°μ‚¬μž…λ‹ˆλ‹€!'}
)
print(f"λŒ“κΈ€ 응닡: {response.status_code}")

# μ„Έμ…˜μ— 인증 μ„€μ • (λͺ¨λ“  μš”μ²­μ— 적용)
session.auth = ('username', 'password')
response = session.get('https://api.example.com/secure')

# μ„Έμ…˜μ— μš”μ²­ ν›… μΆ”κ°€
def logging_hook(response, *args, **kwargs):
    print(f"μš”μ²­ URL: {response.url}")
    print(f"μƒνƒœ μ½”λ“œ: {response.status_code}")
    print(f"응닡 μ‹œκ°„: {response.elapsed.total_seconds()}초")
    return response

session.hooks['response'] = [logging_hook]

# μ„Έμ…˜μ— νƒ€μž„μ•„μ›ƒ μ„€μ •
session.request('GET', 'https://api.example.com/data', timeout=(3.05, 27))  # (μ—°κ²° νƒ€μž„μ•„μ›ƒ, 읽기 νƒ€μž„μ•„μ›ƒ)

# μ»¨ν…μŠ€νŠΈ λ§€λ‹ˆμ €λ‘œ μ„Έμ…˜ μ‚¬μš© (μžλ™μœΌλ‘œ λ¦¬μ†ŒμŠ€ 정리)
with requests.Session() as session:
    session.headers.update({'Authorization': 'Bearer session_token'})
    response = session.get('https://api.example.com/data')
    print(f"응닡: {response.json()}")
    
    # μ„Έμ…˜ λ‚΄μ—μ„œ μ—¬λŸ¬ μš”μ²­ μˆ˜ν–‰
    for page in range(1, 4):
        response = session.get(
            'https://api.example.com/articles',
            params={'page': page}
        )
        articles = response.json().get('articles', [])
        print(f"νŽ˜μ΄μ§€ {page}: {len(articles)} 기사")

# μ„Έμ…˜ μƒνƒœ 확인 및 λ¦¬μ†ŒμŠ€ 정리
print(f"μ„Έμ…˜ μΏ ν‚€: {session.cookies}")
print(f"μ„Έμ…˜ 헀더: {session.headers}")
session.close()  # μ„Έμ…˜ λ¦¬μ†ŒμŠ€ λͺ…μ‹œμ  정리

βœ… νŠΉμ§•:

  • μ„Έμ…˜ μž¬μ‚¬μš©
  • 헀더 μœ μ§€
  • μΏ ν‚€ 관리
  • 인증 정보 보쑴
  • λ¦¬μ†ŒμŠ€ νš¨μœ¨μ„±
  • λ„€νŠΈμ›Œν¬ μ΅œμ ν™”
  • μš”μ²­ ν›… 지원

μ„Έμ…˜ μ‚¬μš©μ˜ 이점:

μ„Έμ…˜μ„ μ‚¬μš©ν•˜μ—¬ HTTP μš”μ²­μ„ 관리할 λ•Œμ˜ μž₯점이닀.

  1. μ—°κ²° μž¬μ‚¬μš©: TCP 연결을 μž¬μ‚¬μš©ν•˜μ—¬ λ„€νŠΈμ›Œν¬ νš¨μœ¨μ„± ν–₯상
  2. μΏ ν‚€ μžλ™ 관리: μ„œλ²„κ°€ μ„€μ •ν•œ μΏ ν‚€λ₯Ό μžλ™μœΌλ‘œ μ €μž₯ν•˜κ³  후속 μš”μ²­μ— 포함
  3. μΌκ΄€λœ 헀더: λͺ¨λ“  μš”μ²­μ— λ™μΌν•œ κΈ°λ³Έ 헀더 적용 κ°€λŠ₯
  4. 인증 μƒνƒœ μœ μ§€: 둜그인 ν›„ 인증 μƒνƒœλ₯Ό μœ μ§€ν•˜λ©° μš”μ²­ κ°€λŠ₯
  5. TCP μ—°κ²° 풀링: HTTP Keep-Aliveλ₯Ό ν†΅ν•œ μ—°κ²° ν’€λ§μœΌλ‘œ μ„±λŠ₯ ν–₯상
  6. μžμ› 관리 κ°„μ†Œν™”: μ»¨ν…μŠ€νŠΈ λ§€λ‹ˆμ €λ₯Ό ν†΅ν•œ μžλ™ λ¦¬μ†ŒμŠ€ 정리 κ°€λŠ₯


4️⃣ μ—λŸ¬ μ²˜λ¦¬μ™€ μž¬μ‹œλ„

import time
from requests.exceptions import RequestException

def make_request_with_retry(url, max_retries=3, delay=1):
    for attempt in range(max_retries):
        try:
            response = requests.get(url)
            response.raise_for_status()
            return response
        except RequestException as e:
            if attempt == max_retries - 1:
                raise e
            time.sleep(delay * (attempt + 1))
    return None

βœ… νŠΉμ§•:

  • μ˜ˆμ™Έ 처리
  • μž¬μ‹œλ„ 둜직
  • μ§€μ—° μ‹œκ°„ μ„€μ •


5️⃣ 비동기 HTTP μš”μ²­

import aiohttp
import asyncio

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        'https://api.example.com/1',
        'https://api.example.com/2',
        'https://api.example.com/3'
    ]
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        return results

βœ… νŠΉμ§•:

  • 비동기 처리
  • λ™μ‹œ μš”μ²­
  • μ„±λŠ₯ μ΅œμ ν™”


6️⃣ μ‹€μš©μ μΈ 예제

class APIClient:
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        
        if api_key:
            self.session.headers.update({
                'Authorization': f'Bearer {api_key}'
            })
    
    def _make_request(self, method, endpoint, **kwargs):
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        response = self.session.request(method, url, **kwargs)
        response.raise_for_status()
        return response.json()
    
    def get(self, endpoint, params=None):
        return self._make_request('GET', endpoint, params=params)

βœ… νŠΉμ§•:

  • 클래슀 기반 κ΅¬ν˜„
  • λ©”μ„œλ“œ 좔상화
  • μž¬μ‚¬μš©μ„±


μ£Όμš” 팁

βœ… λͺ¨λ²” 사둀:

  • νƒ€μž„μ•„μ›ƒ μ„€μ •
  • μ μ ˆν•œ μ—λŸ¬ 처리
  • μ„Έμ…˜ μž¬μ‚¬μš©
  • 비동기 μš”μ²­ ν™œμš©
  • 헀더 관리
  • 인증 처리
  • μž¬μ‹œλ„ 둜직 κ΅¬ν˜„
  • ν”„λ‘μ‹œ μ‚¬μš© κ³ λ €


⚠️ **GitHub.com Fallback** ⚠️