외부 API를 연동하다 보면 항상 마주치는 문제가 있어요. API에서 넘어온 데이터 구조가 우리 앱의 데이터 구조와 전혀 다르다는 점이에요. 예를 들어 외부 API는 id라고 보내는데, 우리 앱에서는 user_id로 써야 하는 경우가 대표적이죠.
이런 문제를 해결하는 가장 효과적인 방법이 바로 중첩 함수를 활용한 자동 변환기를 만드는 거예요. Python의 중첩 함수 기능을 활용하면 변환 로직을 깔끔하게 캡슐화하면서도 재사용이 가능한 변환기를 만들 수 있어요.
기본적인 중첩 함수 변환기 만들기
가장 간단한 형태부터 시작해볼게요. 외부 API에서 받은 필드명을 우리 앱의 필드명으로 매핑하는 변환기예요.
def make_response_transformer(field_map):
"""
field_map: 내부필드명과 외부필드명을 매핑하는 딕셔너리
예시: {"user_id": "id", "user_name": "name"}
"""
def transform(api_response):
# 매핑 규칙에 따라 새로운 딕셔너리 생성
return {
my_key: api_response.get(api_key, None)
for my_key, api_key in field_map.items()
}
return transform
이 변환기의 핵심은 field_map을 클로저로 저장한다는 점이에요. 한 번 생성된 변환기는 동일한 매핑 규칙을 계속 사용할 수 있어요.
실제로 사용하는 방법을 보면 더 이해가 쉬워요.
# 변환 규칙 정의
field_map = {
"user_id": "id",
"user_name": "name",
"user_email": "email"
}
# 변환기 생성
transformer = make_response_transformer(field_map)
# 외부 API 응답 예시
api_response = {
"id": 101,
"name": "홍길동",
"email": "hong@email.com",
"age": 30 # 이 필드는 매핑에 없으므로 무시됨
}
# 변환 실행
internal_data = transformer(api_response)
print(internal_data)
# 출력: {'user_id': 101, 'user_name': '홍길동', 'user_email': 'hong@email.com'}
복잡한 중첩 구조 처리하기
실제 API 응답은 단순한 평면 구조가 아니라 깊게 중첩된 경우가 많아요. 이런 경우를 처리하는 고급 변환기를 만들어봐요.
def make_nested_transformer(path_map):
"""
path_map: 점(.) 표기법으로 경로를 지정하는 매핑
예시: {"user_id": "user.profile.id"}
"""
def get_nested_value(data, path):
# "user.profile.id" 같은 경로를 따라가며 값 추출
keys = path.split('.')
value = data
for key in keys:
if isinstance(value, dict):
value = value.get(key)
if value is None:
return None
else:
return None
return value
def transform(api_response):
result = {}
for my_key, api_path in path_map.items():
result[my_key] = get_nested_value(api_response, api_path)
return result
return transform
# 복잡한 중첩 구조의 API 응답
nested_response = {
"user": {
"profile": {
"id": 123,
"name": "Alice",
"contact": {
"email": "alice@example.com",
"phone": "010-1234-5678"
}
},
"settings": {
"theme": "dark",
"notifications": True
}
}
}
# 경로 기반 매핑 정의
path_map = {
"user_id": "user.profile.id",
"user_name": "user.profile.name",
"email": "user.profile.contact.email",
"theme": "user.settings.theme"
}
# 변환기 생성 및 사용
nested_transformer = make_nested_transformer(path_map)
result = nested_transformer(nested_response)
print(result)
# 출력: {'user_id': 123, 'user_name': 'Alice',
# 'email': 'alice@example.com', 'theme': 'dark'}
데이터 타입 변환과 유효성 검증 추가하기
실무에서는 단순히 필드명만 바꾸는 게 아니라 데이터 타입 변환이나 유효성 검증도 필요해요. 이런 기능을 포함한 더 강력한 변환기를 만들어봐요.
def make_advanced_transformer(field_rules):
"""
field_rules: 각 필드별 변환 규칙을 담은 딕셔너리
{
"내부필드명": {
"source": "외부필드명",
"type": 변환할타입,
"default": 기본값,
"validator": 검증함수 (선택)
}
}
"""
def transform(api_response):
result = {}
for my_key, rule in field_rules.items():
# 원본 값 가져오기
source_key = rule.get("source")
value = api_response.get(source_key)
# 기본값 처리
if value is None:
value = rule.get("default")
# 타입 변환
target_type = rule.get("type")
if target_type and value is not None:
try:
if target_type == str:
value = str(value)
elif target_type == int:
value = int(value)
elif target_type == float:
value = float(value)
elif target_type == bool:
value = bool(value)
except (ValueError, TypeError):
value = rule.get("default")
# 유효성 검증
validator = rule.get("validator")
if validator and callable(validator):
if not validator(value):
value = rule.get("default")
result[my_key] = value
return result
return transform
# 이메일 검증 함수
def is_valid_email(email):
if not email or not isinstance(email, str):
return False
return "@" in email and "." in email.split("@")[1]
# 고급 변환 규칙 정의
advanced_rules = {
"user_id": {
"source": "id",
"type": int,
"default": 0
},
"user_name": {
"source": "name",
"type": str,
"default": "Unknown"
},
"user_email": {
"source": "email",
"type": str,
"default": "",
"validator": is_valid_email
},
"is_active": {
"source": "status",
"type": bool,
"default": False
}
}
# 테스트 데이터
test_response = {
"id": "123", # 문자열이지만 int로 변환됨
"name": "김철수",
"email": "invalid-email", # 유효하지 않은 이메일
"status": 1 # 1은 True로 변환됨
}
# 변환기 사용
advanced_transformer = make_advanced_transformer(advanced_rules)
result = advanced_transformer(test_response)
print(result)
# 출력: {'user_id': 123, 'user_name': '김철수',
# 'user_email': '', 'is_active': True}
리스트 데이터 일괄 변환하기
API가 여러 개의 아이템을 리스트로 반환하는 경우도 많아요. 이런 경우를 위한 일괄 변환 기능을 추가해봐요.
def make_batch_transformer(single_transformer):
"""
단일 변환기를 받아서 리스트 전체를 변환하는 변환기 생성
"""
def batch_transform(api_response_list):
# 리스트가 아니면 단일 아이템으로 처리
if not isinstance(api_response_list, list):
return single_transformer(api_response_list)
# 리스트의 각 아이템을 변환
return [single_transformer(item) for item in api_response_list]
return batch_transform
# 기본 변환기 생성
basic_transformer = make_response_transformer({
"product_id": "id",
"product_name": "title",
"price": "cost"
})
# 일괄 변환기로 감싸기
batch_transformer = make_batch_transformer(basic_transformer)
# 여러 상품 데이터
products_response = [
{"id": 1, "title": "노트북", "cost": 1200000},
{"id": 2, "title": "마우스", "cost": 30000},
{"id": 3, "title": "키보드", "cost": 80000}
]
# 일괄 변환
transformed_products = batch_transformer(products_response)
for product in transformed_products:
print(product)
# 출력:
# {'product_id': 1, 'product_name': '노트북', 'price': 1200000}
# {'product_id': 2, 'product_name': '마우스', 'price': 30000}
# {'product_id': 3, 'product_name': '키보드', 'price': 80000}
실시간 API 연동에 적용하기
이제 실제 API 호출과 함께 사용하는 완전한 예제를 만들어봐요. macOS의 터미널에서 바로 실행할 수 있어요.
import requests
from datetime import datetime
def create_api_client(base_url, transformer):
"""
API 클라이언트와 변환기를 결합한 래퍼 함수 생성
"""
def fetch_and_transform(endpoint, params=None):
try:
# API 호출
response = requests.get(f"{base_url}/{endpoint}", params=params)
response.raise_for_status()
# JSON 파싱
data = response.json()
# 변환 적용
if isinstance(data, list):
# 리스트인 경우 각 항목 변환
return [transformer(item) for item in data]
else:
# 단일 객체인 경우
return transformer(data)
except requests.exceptions.RequestException as e:
print(f"API 호출 오류: {e}")
return None
except ValueError as e:
print(f"JSON 파싱 오류: {e}")
return None
return fetch_and_transform
# GitHub API용 변환 규칙 (예시)
github_transformer = make_response_transformer({
"repo_id": "id",
"repo_name": "name",
"owner": "owner.login",
"stars": "stargazers_count",
"language": "language",
"url": "html_url"
})
# 실제 사용 예시 (GitHub API는 인증 없이도 사용 가능)
# api_client = create_api_client("https://api.github.com", github_transformer)
# repos = api_client("users/torvalds/repos", params={"per_page": 5})
# 로컬 테스트용 모의 데이터
mock_github_response = {
"id": 12345,
"name": "linux",
"owner": {"login": "torvalds"},
"stargazers_count": 50000,
"language": "C",
"html_url": "https://github.com/torvalds/linux"
}
# 변환 테스트
transformed = github_transformer(mock_github_response)
print(f"변환 결과: {transformed}")
# 출력: 변환 결과: {'repo_id': 12345, 'repo_name': 'linux', ...}
성능 최적화와 캐싱 전략
실시간 API 응답 변환에서 성능은 매우 중요해요. 변환 로직을 최적화하고 캐싱을 적용하는 방법을 알아봐요.
from functools import lru_cache
import time
def make_cached_transformer(field_map, cache_size=128):
"""
LRU 캐시를 적용한 변환기 생성
동일한 입력에 대해 변환 결과를 캐싱
"""
# 딕셔너리를 튜플로 변환하는 헬퍼 함수
def dict_to_tuple(d):
return tuple(sorted(d.items()))
@lru_cache(maxsize=cache_size)
def cached_transform(api_response_tuple):
# 튜플을 다시 딕셔너리로 변환
api_response = dict(api_response_tuple)
# 실제 변환 수행
return {
my_key: api_response.get(api_key, None)
for my_key, api_key in field_map.items()
}
def transform(api_response):
# 딕셔너리를 튜플로 변환해서 캐시 키로 사용
response_tuple = dict_to_tuple(api_response)
return cached_transform(response_tuple)
# 캐시 상태 확인용 메서드 추가
transform.cache_info = cached_transform.cache_info
transform.cache_clear = cached_transform.cache_clear
return transform
# 성능 테스트
field_map = {"user_id": "id", "user_name": "name"}
cached_transformer = make_cached_transformer(field_map)
# 동일한 데이터로 여러 번 변환
test_data = {"id": 1, "name": "테스트"}
# 첫 번째 호출 (캐시 미스)
start = time.time()
result1 = cached_transformer(test_data)
first_call_time = time.time() - start
# 두 번째 호출 (캐시 히트)
start = time.time()
result2 = cached_transformer(test_data)
second_call_time = time.time() - start
print(f"첫 번째 호출 시간: {first_call_time:.6f}초")
print(f"두 번째 호출 시간: {second_call_time:.6f}초")
print(f"캐시 정보: {cached_transformer.cache_info()}")
에러 처리와 로깅 추가하기
실제 서비스에서는 예외 상황을 우아하게 처리하고 디버깅을 위한 로그를 남기는 것이 중요해요.
import logging
from datetime import datetime
# 로거 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def make_robust_transformer(field_map, strict=False):
"""
에러 처리와 로깅 기능이 포함된 견고한 변환기
strict=True: 에러 발생 시 예외 발생
strict=False: 에러 발생 시 기본값 반환
"""
def transform(api_response):
start_time = datetime.now()
result = {}
errors = []
# 입력 검증
if not isinstance(api_response, dict):
error_msg = f"Invalid input type: {type(api_response)}"
logger.error(error_msg)
if strict:
raise TypeError(error_msg)
return {}
# 각 필드 변환
for my_key, api_key in field_map.items():
try:
value = api_response.get(api_key)
result[my_key] = value
if value is None:
logger.warning(f"Missing field: {api_key}")
except Exception as e:
error_msg = f"Error transforming {api_key}: {str(e)}"
logger.error(error_msg)
errors.append(error_msg)
if strict:
raise
else:
result[my_key] = None
# 변환 완료 로그
elapsed_time = (datetime.now() - start_time).total_seconds()
logger.info(f"Transformation completed in {elapsed_time:.4f}s")
if errors:
logger.warning(f"Transformation had {len(errors)} errors")
return result
return transform
# 사용 예시
transformer = make_robust_transformer({
"user_id": "id",
"user_name": "name",
"user_level": "level"
}, strict=False)
# 일부 필드가 누락된 응답
incomplete_response = {
"id": 100,
"name": "테스트유저"
# "level" 필드가 없음
}
result = transformer(incomplete_response)
print(f"변환 결과: {result}")
# 로그 출력과 함께 변환 결과 반환
중첩 함수를 활용한 API 응답 변환기는 외부 API와의 통합을 깔끔하게 처리하는 강력한 패턴이에요. 클로저를 통해 설정을 유지하면서도 재사용 가능한 변환 로직을 만들 수 있고, 필요에 따라 캐싱, 에러 처리, 로깅 등의 기능을 쉽게 추가할 수 있어요.
특히 여러 외부 API를 사용하는 서비스에서는 각 API별로 맞춤형 변환기를 만들어 관리하면 코드의 일관성과 유지보수성을 크게 향상시킬 수 있어요.