pip install requests

import requests

# GET 요청 보내기
response = requests.get('https://jsonplaceholder.typicode.com/posts/1')

# 상태 코드 확인
print(response.status_code)  # 200

# JSON 응답 파싱
data = response.json()
print(data)
# {'userId': 1, 'id': 1, 'title': '...', 'body': '...'}

response = requests.get('https://jsonplaceholder.typicode.com/posts/1')

print(response.status_code)    # 200 (상태 코드)
print(response.headers)        # 응답 헤더 딕셔너리
print(response.text)           # 응답 본문 (문자열)
print(response.json())         # JSON 파싱 결과 (딕셔너리)
print(response.content)        # 응답 본문 (바이트)
print(response.url)            # 최종 요청 URL
print(response.elapsed)        # 응답 시간
print(response.encoding)       # 인코딩 방식

# 성공 여부 확인 (200~299면 True)
print(response.ok)             # True

import requests

# 방법 1: URL에 직접 붙이기 (비추천)
response = requests.get('https://api.example.com/search?q=python&page=1&limit=10')

# 방법 2: params 딕셔너리 사용 (추천! ✅)
params = {
    'q': 'python',
    'page': 1,
    'limit': 10,
    'sort': 'latest'
}
response = requests.get('https://api.example.com/search', params=params)

# 실제 요청 URL 확인
print(response.url)
# https://api.example.com/search?q=python&page=1&limit=10&sort=latest

import requests

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

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

print(response.json())

import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth

# 1. Basic 인증 (아이디/비밀번호)
response = requests.get(
    'https://api.example.com/data',
    auth=HTTPBasicAuth('username', 'password')
)
# 또는 간단하게
response = requests.get(
    'https://api.example.com/data',
    auth=('username', 'password')
)

# 2. API Key 인증 (헤더에 넣기)
headers = {'X-API-Key': 'your_api_key_here'}
response = requests.get('https://api.example.com/data', headers=headers)

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

import os
import requests

# 환경변수에서 API 키 가져오기 (안전한 방법 ✅)
api_key = os.environ.get('MY_API_KEY')

headers = {'Authorization': f'Bearer {api_key}'}
response = requests.get('https://api.example.com/data', headers=headers)

import requests

# JSON 데이터 보내기
url = 'https://jsonplaceholder.typicode.com/posts'

payload = {
    'title': '파이썬으로 API 연동하기',
    'body': '오늘 배운 내용을 정리해봤어요!',
    'userId': 1
}

# json 파라미터 사용 (Content-Type 자동 설정)
response = requests.post(url, json=payload)

print(response.status_code)  # 201 Created
print(response.json())
# {'id': 101, 'title': '파이썬으로 API 연동하기', ...}

import requests

# application/x-www-form-urlencoded 형식
form_data = {
    'username': 'pythonlover',
    'password': 'secret123'
}
response = requests.post('https://api.example.com/login', data=form_data)

# multipart/form-data (파일 업로드)
files = {
    'profile_image': ('avatar.png', open('avatar.png', 'rb'), 'image/png')
}
response = requests.post('https://api.example.com/upload', files=files)

import requests

# PUT: 전체 리소스 교체
response = requests.put(
    'https://jsonplaceholder.typicode.com/posts/1',
    json={
        'id': 1,
        'title': '수정된 제목',
        'body': '수정된 내용',
        'userId': 1
    }
)

# PATCH: 일부 필드만 수정
response = requests.patch(
    'https://jsonplaceholder.typicode.com/posts/1',
    json={'title': '제목만 수정'}
)

print(response.status_code)  # 200
print(response.json())

import requests

response = requests.delete('https://jsonplaceholder.typicode.com/posts/1')
print(response.status_code)  # 200 또는 204 No Content

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

def fetch_data(url, params=None):
    try:
        response = requests.get(
            url,
            params=params,
            timeout=10  # 10초 타임아웃 설정
        )
        
        # HTTP 에러 (4xx, 5xx) 발생 시 예외 raise
        response.raise_for_status()
        
        return response.json()
    
    except Timeout:
        print("⏰ 요청 시간이 초과됐어요. 서버가 느린가봐요.")
        return None
    
    except ConnectionError:
        print("🔌 연결 실패! 인터넷 연결을 확인해주세요.")
        return None
    
    except HTTPError as e:
        status_code = e.response.status_code
        if status_code == 401:
            print("🔑 인증 실패! API 키를 확인해주세요.")
        elif status_code == 403:
            print("🚫 접근 권한이 없어요.")
        elif status_code == 404:
            print("🔍 요청한 리소스를 찾을 수 없어요.")
        elif status_code == 429:
            print("⏳ 요청이 너무 많아요. 잠시 후 다시 시도해주세요.")
        elif status_code >= 500:
            print(f"💥 서버 에러 발생! 상태 코드: {status_code}")
        return None
    
    except RequestException as e:
        print(f"❌ 알 수 없는 오류: {e}")
        return None

# 사용
data = fetch_data('https://jsonplaceholder.typicode.com/posts/1')
if data:
    print(data)

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

def create_session_with_retry():
    session = requests.Session()
    
    retry_strategy = Retry(
        total=3,              # 최대 3번 재시도
        backoff_factor=1,     # 재시도 간격: 1초, 2초, 4초...
        status_forcelist=[429, 500, 502, 503, 504],  # 재시도할 상태 코드
        allowed_methods=["GET", "POST"]  # 재시도할 메서드
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

# 사용
session = create_session_with_retry()
response = session.get('https://api.example.com/data', timeout=10)
print(response.json())

import requests

# Session 생성
session = requests.Session()

# 공통 헤더 설정 (모든 요청에 자동 적용)
session.headers.update({
    'Authorization': 'Bearer your_token',
    'Content-Type': 'application/json',
    'User-Agent': 'MyApp/1.0'
})

# 공통 파라미터 설정
session.params = {'api_version': 'v2'}

# 이제 모든 요청에 위 설정이 자동 적용돼!
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': '좋아요!'})

# Session 종료 (with 문 사용 권장)
session.close()

# 더 좋은 방법: with 문으로 자동 종료
with requests.Session() as session:
    session.headers.update({'Authorization': 'Bearer your_token'})
    
    users = session.get('https://api.example.com/users').json()
    posts = session.get('https://api.example.com/posts').json()
    
    print(f"유저 수: {len(users)}")
    print(f"포스트 수: {len(posts)}")

with requests.Session() as session:
    # 로그인
    login_response = session.post(
        'https://example.com/login',
        data={'username': 'user', 'password': 'pass'}
    )
    
    # 로그인 후 쿠키가 자동으로 세션에 저장됨
    # 이후 요청에 자동으로 쿠키가 포함됨
    profile = session.get('https://example.com/profile')
    print(profile.json())

import requests
import os

def get_weather(city: str, api_key: str) -> dict:
    """도시 이름으로 현재 날씨 조회"""
    
    url = "https://api.openweathermap.org/data/2.5/weather"
    params = {
        'q': city,
        'appid': api_key,
        'units': 'metric',  # 섭씨 온도
        'lang': 'kr'        # 한국어 설명
    }
    
    try:
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        
        return {
            '도시': data['name'],
            '날씨': data['weather'][0]['description'],
            '온도': f"{data['main']['temp']}°C",
            '체감온도': f"{data['main']['feels_like']}°C",
            '습도': f"{data['main']['humidity']}%",
            '풍속': f"{data['wind']['speed']}m/s"
        }
    
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 401:
            return {'에러': 'API 키가 유효하지 않아요!'}
        elif e.response.status_code == 404:
            return {'에러': f'"{city}" 도시를 찾을 수 없어요!'}
        return {'에러': str(e)}
    
    except Exception as e:
        return {'에러': str(e)}

# 사용 예시
api_key = os.environ.get('OPENWEATHER_API_KEY')
weather = get_weather('Seoul', api_key)

for key, value in weather.items():
    print(f"{key}: {value}")

# 출력 예시:
# 도시: Seoul
# 날씨: 맑음
# 온도: 22.5°C
# 체감온도: 21.8°C
# 습도: 65%
# 풍속: 3.2m/s

import requests

class GitHubClient:
    """GitHub API 클라이언트"""
    
    BASE_URL = "https://api.github.com"
    
    def __init__(self, token: str = None):
        self.session = requests.Session()
        self.session.headers.update({
            'Accept': 'application/vnd.github.v3+json',
            'User-Agent': 'MyGitHubApp/1.0'
        })
        
        # 토큰이 있으면 인증 헤더 추가
        if token:
            self.session.headers.update({
                'Authorization': f'token {token}'
            })
    
    def get_user(self, username: str) -> dict:
        """사용자 정보 조회"""
        response = self.session.get(f"{self.BASE_URL}/users/{username}")
        response.raise_for_status()
        return response.json()
    
    def get_repos(self, username: str, sort: str = 'updated') -> list:
        """사용자 레포지토리 목록 조회"""
        params = {
            'sort': sort,
            'per_page': 10
        }
        response = self.session.get(
            f"{self.BASE_URL}/users/{username}/repos",
            params=params
        )
        response.raise_for_status()
        return response.json()
    
    def get_repo_info(self, owner: str, repo: str) -> dict:
        """특정 레포지토리 정보 조회"""
        response = self.session.get(f"{self.BASE_URL}/repos/{owner}/{repo}")
        response.raise_for_status()
        return response.json()
    
    def create_issue(self, owner: str, repo: str, title: str, body: str) -> dict:
        """이슈 생성 (토큰 필요)"""
        payload = {'title': title, 'body': body}
        response = self.session.post(
            f"{self.BASE_URL}/repos/{owner}/{repo}/issues",
            json=payload
        )
        response.raise_for_status()
        return response.json()
    
    def __del__(self):
        self.session.close()

# 사용 예시
client = GitHubClient()  # 공개 API는 토큰 없이도 가능

# 사용자 정보 조회
user = client.get_user('torvalds')
print(f"이름: {user['name']}")
print(f"팔로워: {user['followers']:,}명")
print(f"공개 레포: {user['public_repos']}개")

# 레포지토리 목록
repos = client.get_repos('torvalds')
print("\n최근 업데이트된 레포지토리:")
for repo in repos[:5]:
    stars = repo['stargazers_count']
    print(f"  ⭐ {repo['name']} ({stars:,} stars)")

import requests
import os

def search_address(query: str, api_key: str) -> list:
    """카카오 주소 검색 API"""
    
    url = "https://dapi.kakao.com/v2/local/search/address.json"
    headers = {
        'Authorization': f'KakaoAK {api_key}'
    }
    params = {
        'query': query,
        'size': 5  # 최대 5개 결과
    }
    
    response = requests.get(url, headers=headers, params=params, timeout=10)
    response.raise_for_status()
    
    data = response.json()
    results = []
    
    for doc in data['documents']:
        results.append({
            '주소': doc['address_name'],
            '위도': doc['y'],
            '경도': doc['x'],
            '주소타입': doc['address_type']
        })
    
    return results

def search_keyword(keyword: str, api_key: str, category: str = None) -> list:
    """카카오 키워드 장소 검색 API"""
    
    url = "https://dapi.kakao.com/v2/local/search/keyword.json"
    headers = {'Authorization': f'KakaoAK {api_key}'}
    params = {
        'query': keyword,
        'size': 5
    }
    
    if category:
        params['category_group_code'] = category
    
    response = requests.get(url, headers=headers, params=params, timeout=10)
    response.raise_for_status()
    
    data = response.json()
    results = []
    
    for place in data['documents']:
        results.append({
            '장소명': place['place_name'],
            '주소': place['road_address_name'],
            '전화번호': place.get('phone', '없음'),
            '카테고리': place['category_name']
        })
    
    return results

# 사용 예시
kakao_key = os.environ.get('KAKAO_REST_API_KEY')

# 주소 검색
addresses = search_address('강남구 테헤란로 152', kakao_key)
for addr in addresses:
    print(f"📍 {addr['주소']} (위도: {addr['위도']}, 경도: {addr['경도']})")

# 카페 검색
cafes = search_keyword('강남역 카페', kakao_key, 'CE7')
for cafe in cafes:
    print(f"☕ {cafe['장소명']} - {cafe['주소']}")

import requests
import time
import logging
from typing import Optional, Dict, Any
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class BaseAPIClient:
    """재사용 가능한 API 클라이언트 베이스 클래스"""
    
    def __init__(
        self,
        base_url: str,
        api_key: str = None,
        timeout: int = 10,
        max_retries: int = 3
    ):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = self._create_session(api_key, max_retries)
    
    def _create_session(self, api_key: str, max_retries: int) -> requests.Session:
        """재시도 로직이 포함된 세션 생성"""
        session = requests.Session()
        
        # 기본 헤더 설정
        session.headers.update({
            'Content-Type': 'application/json',
            'Accept': 'application/json'
        })
        
        # API 키 설정
        if api_key:
            session.headers.update({'Authorization': f'Bearer {api_key}'})
        
        # 재시도 전략 설정
        retry = Retry(
            total=max_retries,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504]
        )
        adapter = HTTPAdapter(max_retries=retry)
        session.mount('https://', adapter)
        session.mount('http://', adapter)
        
        return session
    
    def _request(
        self,
        method: str,
        endpoint: str,
        **kwargs
    ) -> Optional[Dict[str, Any]]:
        """공통 요청 메서드"""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        
        try:
            logger.info(f"[{method.upper()}] {url}")
            start_time = time.time()
            
            response = self.session.request(
                method,
                url,
                timeout=self.timeout,
                **kwargs
            )
            
            elapsed = time.time() - start_time
            logger.info(f"응답: {response.status_code} ({elapsed:.2f}초)")
            
            response.raise_for_status()
            
            # 응답 본문이 없는 경우 처리 (204 No Content 등)
            if response.status_code == 204 or not response.content:
                return {}
            
            return response.json()
        
        except requests.exceptions.HTTPError as e:
            logger.error(f"HTTP 에러: {e.response.status_code} - {e.response.text}")
            raise
        
        except requests.exceptions.Timeout:
            logger.error(f"타임아웃: {url}")
            raise
        
        except requests.exceptions.RequestException as e:
            logger.error(f"요청 실패: {e}")
            raise
    
    def get(self, endpoint: str, params: dict = None) -> dict:
        return self._request('GET', endpoint, params=params)
    
    def post(self, endpoint: str, data: dict = None) -> dict:
        return self._request('POST', endpoint, json=data)
    
    def put(self, endpoint: str, data: dict = None) -> dict:
        return self._request('PUT', endpoint, json=data)
    
    def patch(self, endpoint: str, data: dict = None) -> dict:
        return self._request('PATCH', endpoint, json=data)
    
    def delete(self, endpoint: str) -> dict:
        return self._request('DELETE', endpoint)
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.session.close()


# 실제 서비스 클라이언트 구현 예시
class TodoAPIClient(BaseAPIClient):
    """할 일 관리 API 클라이언트"""
    
    def __init__(self, api_key: str = None):
        super().__init__(
            base_url='https://jsonplaceholder.typicode.com',
            api_key=api_key
        )
    
    def get_todos(self, user_id: int = None) -> list:
        """할 일 목록 조회"""
        params = {'userId': user_id} if user_id else None
        return self.get('/todos', params=params)
    
    def get_todo(self, todo_id: int) -> dict:
        """특정 할 일 조회"""
        return self.get(f'/todos/{todo_id}')
    
    def create_todo(self, title: str, user_id: int) -> dict:
        """할 일 생성"""
        return self.post('/todos', data={
            'title': title,
            'userId': user_id,
            'completed': False
        })
    
    def complete_todo(self, todo_id: int) -> dict:
        """할 일 완료 처리"""
        return self.patch(f'/todos/{todo_id}', data={'completed': True})
    
    def delete_todo(self, todo_id: int) -> dict:
        """할 일 삭제"""
        return self.delete(f'/todos/{todo_id}')


# 사용 예시
with TodoAPIClient() as client:
    # 할 일 목록 조회
    todos = client.get_todos(user_id=1)
    print(f"총 {len(todos)}개의 할 일")
    
    # 새 할 일 생성
    new_todo = client.create_todo('파이썬 API 공부하기', user_id=1)
    print(f"생성됨: {new_todo}")
    
    # 완료 처리
    completed = client.complete_todo(1)
    print(f"완료: {completed}")

import requests
import time

def fetch_all_pages(base_url: str, params: dict = None, max_pages: int = 10) -> list:
    """모든 페이지의 데이터를 가져오는 함수"""
    
    all_data = []
    params = params or {}
    params['page'] = 1
    params['per_page'] = 100  # 페이지당 최대 항목 수
    
    for page_num in range(1, max_pages + 1):
        params['page'] = page_num
        
        response = requests.get(base_url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        
        # 데이터가 없으면 종료
        if not data:
            print(f"✅ 총 {page_num - 1}페이지 완료!")
            break
        
        all_data.extend(data)
        print(f"📄 {page_num}페이지 로드: {len(data)}개 항목")
        
        # API 레이트 리밋 방지: 요청 간 딜레이
        time.sleep(0.5)
        
        # 마지막 페이지 감지 (Link 헤더 방식)
        link_header = response.headers.get('Link', '')
        if 'rel="next"' not in link_header:
            print(f"✅ 마지막 페이지 도달!")
            break
    
    return all_data


# GitHub API 페이지네이션 예시
def get_all_repos(username: str, token: str = None) -> list:
    """사용자의 모든 레포지토리 가져오기"""
    
    headers = {'Accept': 'application/vnd.github.v3+json'}
    if token:
        headers['Authorization'] = f'token {token}'
    
    all_repos = []
    page = 1
    
    while True:
        response = requests.get(
            f'https://api.github.com/users/{username}/repos',
            headers=headers,
            params={'page': page, 'per_page': 30, 'sort': 'updated'},
            timeout=10
        )
        response.raise_for_status()
        
        repos = response.json()
        
        if not repos:
            break
        
        all_repos.extend(repos)
        
        # Link 헤더로 다음 페이지 확인
        if 'rel="next"' not in response.headers.get('Link', ''):
            break
        
        page += 1
        time.sleep(0.3)  # 레이트 리밋 방지
    
    return all_repos

# 사용
repos = get_all_repos('octocat')
print(f"총 레포지토리: {len(repos)}개")
for repo in repos:
    print(f"  📁 {repo['name']} ⭐{repo['stargazers_count']}")

import requests
import os

# 카카오 OAuth 2.0 예시

# 1단계: 인증 URL 생성 (사용자를 이 URL로 리다이렉트)
def get_kakao_auth_url():
    client_id = os.environ.get('KAKAO_CLIENT_ID')
    redirect_uri = 'https://myapp.com/callback'
    
    return (
        f"https://kauth.kakao.com/oauth/authorize"
        f"?client_id={client_id}"
        f"&redirect_uri={redirect_uri}"
        f"&response_type=code"
    )

# 2단계: 인증 코드로 액세스 토큰 발급
def get_kakao_token(auth_code: str) -> dict:
    client_id = os.environ.get('KAKAO_CLIENT_ID')
    client_secret = os.environ.get('KAKAO_CLIENT_SECRET')
    redirect_uri = 'https://myapp.com/callback'
    
    response = requests.post(
        'https://kauth.kakao.com/oauth/token',
        data={
            'grant_type': 'authorization_code',
            'client_id': client_id,
            'client_secret': client_secret,
            'redirect_uri': redirect_uri,
            'code': auth_code
        }
    )
    response.raise_for_status()
    return response.json()
    # {'access_token': '...', 'token_type': 'bearer', 'expires_in': 21599, ...}

# 3단계: 액세스 토큰으로 사용자 정보 조회
def get_kakao_user_info(access_token: str) -> dict:
    response = requests.get(
        'https://kapi.kakao.com/v2/user/me',
        headers={'Authorization': f'Bearer {access_token}'}
    )
    response.raise_for_status()
    return response.json()

# 사용 흐름
auth_url = get_kakao_auth_url()
print(f"이 URL로 사용자를 리다이렉트: {auth_url}")

# 사용자가 로그인 후 콜백으로 code 파라미터가 옴
auth_code = "받은_인증_코드"
token_data = get_kakao_token(auth_code)
access_token = token_data['access_token']

user_info = get_kakao_user_info(access_token)
print(f"사용자 ID: {user_info['id']}")
print(f"닉네임: {user_info['kakao_account']['profile']['nickname']}")

import requests

# 연결 타임아웃 + 읽기 타임아웃 분리 설정
response = requests.get(
    'https://api.example.com/data',
    timeout=(3.05, 27)  # (연결 타임아웃, 읽기 타임아웃) 초
)

# 단일 타임아웃 (연결+읽기 합산)
response = requests.get('https://api.example.com/data', timeout=10)

import requests

proxies = {
    'http': 'http://proxy.example.com:8080',
    'https': 'http://proxy.example.com:8080'
}

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

import requests

# SSL 인증서 검증 비활성화 (개발 환경에서만! 프로덕션에서는 절대 금지)
response = requests.get('https://api.example.com/data', verify=False)

# 커스텀 CA 인증서 사용
response = requests.get(
    'https://api.example.com/data',
    verify='/path/to/ca-bundle.crt'
)

import requests

def download_file(url: str, filename: str):
    """대용량 파일을 청크 단위로 다운로드"""
    
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        
        total_size = int(response.headers.get('content-length', 0))
        downloaded = 0
        
        with open(filename, 'wb') as f:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:  # 빈 청크 필터링
                    f.write(chunk)
                    downloaded += len(chunk)
                    
                    if total_size:
                        progress = (downloaded / total_size) * 100
                        print(f"\r다운로드: {progress:.1f}%", end='', flush=True)
        
        print(f"\n✅ 다운로드 완료: {filename}")

# 사용
download_file(
    'https://example.com/large-file.zip',
    'downloaded-file.zip'
)

import unittest
from unittest.mock import patch, MagicMock
import requests

# 테스트할 함수
def get_user_info(user_id: int) -> dict:
    response = requests.get(f'https://api.example.com/users/{user_id}')
    response.raise_for_status()
    return response.json()

# 테스트 코드
class TestAPIClient(unittest.TestCase):
    
    @patch('requests.get')
    def test_get_user_success(self, mock_get):
        """성공 케이스 테스트"""
        # Mock 응답 설정
        mock_response = MagicMock()
        mock_response.status_code = 200
        mock_response.json.return_value = {
            'id': 1,
            'name': '홍길동',
            'email': 'hong@example.com'
        }
        mock_get.return_value = mock_response
        
        # 함수 호출
        result = get_user_info(1)
        
        # 검증
        self.assertEqual(result['name'], '홍길동')
        mock_get.assert_called_once_with('https://api.example.com/users/1')
    
    @patch('requests.get')
    def test_get_user_not_found(self, mock_get):
        """404 에러 케이스 테스트"""
        mock_response = MagicMock()
        mock_response.status_code = 404
        mock_response.raise_for_status.side_effect = requests.exceptions.HTTPError(
            response=mock_response
        )
        mock_get.return_value = mock_response
        
        # 예외 발생 확인
        with self.assertRaises(requests.exceptions.HTTPError):
            get_user_info(999)
    
    @patch('requests.get')
    def test_timeout(self, mock_get):
        """타임아웃 케이스 테스트"""
        mock_get.side_effect = requests.exceptions.Timeout()
        
        with self.assertRaises(requests.exceptions.Timeout):
            get_user_info(1)

if __name__ == '__main__':
    unittest.main()

import requests
import time
from functools import wraps

def rate_limited(max_calls: int, period: float):
    """레이트 리밋 데코레이터"""
    calls = []
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            
            # 기간 내 호출 기록 필터링
            calls[:] = [c for c in calls if now - c < period]
            
            if len(calls) >= max_calls:
                # 대기 시간 계산
                sleep_time = period - (now - calls[0])
                if sleep_time > 0:
                    print(f"⏳ 레이트 리밋 대기: {sleep_time:.2f}초")
                    time.sleep(sleep_time)
            
            calls.append(time.time())
            return func(*args, **kwargs)
        
        return wrapper
    return decorator

# 초당 최대 5번 호출 제한
@rate_limited(max_calls=5, period=1.0)
def call_api(endpoint: str) -> dict:
    response = requests.get(endpoint, timeout=10)
    response.raise_for_status()
    return response.json()

# 응답 헤더에서 레이트 리밋 정보 읽기
def smart_request(url: str) -> dict:
    response = requests.get(url, timeout=10)
    
    # GitHub API 레이트 리밋 헤더 예시
    remaining = int(response.headers.get('X-RateLimit-Remaining', 999))
    reset_time = int(response.headers.get('X-RateLimit-Reset', 0))
    
    if response.status_code == 429:
        # 리셋 시간까지 대기
        wait_time = reset_time - time.time() + 1
        print(f"⏳ 레이트 리밋 초과! {wait_time:.0f}초 대기...")
        time.sleep(max(wait_time, 0))
        return smart_request(url)  # 재시도
    
    if remaining < 10:
        print(f"⚠️ 남은 요청 횟수: {remaining}회")
    
    response.raise_for_status()
    return response.json()

import requests
import time
import os
from datetime import datetime
from typing import Optional

class ExchangeRateMonitor:
    """환율 모니터링 클라이언트"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = f"https://v6.exchangerate-api.com/v6/{api_key}"
        self.session = requests.Session()
        self.session.headers.update({'Accept': 'application/json'})
    
    def get_rate(self, from_currency: str, to_currency: str) -> Optional[float]:
        """현재 환율 조회"""
        try:
            response = self.session.get(
                f"{self.base_url}/pair/{from_currency}/{to_currency}",
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            if data['result'] == 'success':
                return data['conversion_rate']
            return None
        
        except Exception as e:
            print(f"❌ 환율 조회 실패: {e}")
            return None
    
    def get_all_rates(self, base_currency: str = 'USD') -> dict:
        """기준 통화 대비 모든 환율 조회"""
        try:
            response = self.session.get(
                f"{self.base_url}/latest/{base_currency}",
                timeout=10
            )
            response.raise_for_status()
            data = response.json()
            
            if data['result'] == 'success':
                return data['conversion_rates']
            return {}
        
        except Exception as e:
            print(f"❌ 환율 조회 실패: {e}")
            return {}
    
    def send_slack_alert(self, webhook_url: str, message: str):
        """슬랙 웹훅으로 알림 전송"""
        try:
            response = requests.post(
                webhook_url,
                json={'text': message},
                timeout=10
            )
            response.raise_for_status()
            print("✅ 슬랙 알림 전송 완료!")
        except Exception as e:
            print(f"❌ 슬랙 알림 실패: {e}")
    
    def monitor(
        self,
        from_currency: str,
        to_currency: str,
        target_rate: float,
        condition: str = 'below',  # 'below' 또는 'above'
        check_interval: int = 60,  # 초
        webhook_url: str = None
    ):
        """환율 모니터링 시작"""
        print(f"🔍 환율 모니터링 시작!")
        print(f"   {from_currency} → {to_currency}")
        print(f"   목표: {condition} {target_rate}")
        print(f"   체크 간격: {check_interval}초")
        print("-" * 40)
        
        while True:
            rate = self.get_rate(from_currency, to_currency)
            
            if rate is None:
                print("⚠️ 환율 조회 실패, 재시도...")
                time.sleep(check_interval)
                continue
            
            now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
            print(f"[{now}] {from_currency}/{to_currency} = {rate:.4f}")
            
            # 목표 환율 도달 확인
            triggered = False
            if condition == 'below' and rate <= target_rate:
                triggered = True
                msg = f"🎯 목표 환율 도달!\n{from_currency}/{to_currency} = {rate:.4f}\n목표: {target_rate} 이하"
            elif condition == 'above' and rate >= target_rate:
                triggered = True
                msg = f"🎯 목표 환율 도달!\n{from_currency}/{to_currency} = {rate:.4f}\n목표: {target_rate} 이상"
            
            if triggered:
                print(f"\n🚨 {msg}")
                if webhook_url:
                    self.send_slack_alert(webhook_url, msg)
                break
            
            time.sleep(check_interval)
    
    def __del__(self):
        self.session.close()

# 사용 예시
api_key = os.environ.get('EXCHANGE_RATE_API_KEY')
monitor = ExchangeRateMonitor(api_key)

# 현재 환율 확인
usd_krw = monitor.get_rate('USD', 'KRW')
print(f"현재 USD/KRW: {usd_krw:.2f}원")

# 주요 통화 환율 조회
rates = monitor.get_all_rates('USD')
currencies = ['KRW', 'JPY', 'EUR', 'GBP', 'CNY']
print("\n📊 주요 통화 환율 (USD 기준):")
for currency in currencies:
    if currency in rates:
        print(f"  {currency}: {rates[currency]:.4f}")

# 환율 모니터링 (1달러 = 1300원 이하가 되면 알림)
# monitor.monitor(
#     from_currency='USD',
#     to_currency='KRW',
#     target_rate=1300,
#     condition='below',
#     check_interval=300,  # 5분마다 체크
#     webhook_url=os.environ.get('SLACK_WEBHOOK_URL')
# )