어댑터 패턴 개요

어댑터(Adapter) 패턴은 서로 다른 인터페이스를 가진 클래스들이 협업할 수 있도록 중간에서 인터페이스를 변환해주는 구조적 디자인 패턴입니다. 하위 시스템이나 외부 서비스의 API가 변경되더라도 클라이언트 코드를 보호하는 역할을 합니다.

일반적으로 다음과 같은 상황에서 유용합니다.

• 서드파티 라이브러리의 메서드 시그니처가 프로젝트 규약과 다를 때
• 레거시 모듈과 새로운 모듈을 동시에 사용해야 할 때
• 여러 벤더의 비슷한 기능을 하나의 추상화 인터페이스로 제공해야 할 때

Python에서의 구현 방식

Python에서는 다중상속을 활용한 클래스适格과 합성을 활용한 객체适格 두 가지 방식으로 구현할 수 있습니다. 일반적으로 객체适格이 유연성과 테스트 용이성 측면에서 권장됩니다.

클래스适格 예시

class NewLogger:
    def write_log(self, message: str) -> None:
        print(f"[NEW] {message}")

class LegacyLogger:
    def log_legacy(self, text: str) -> None:
        print(f"[LEGACY] {text}")

class LoggerAdapter(NewLogger, LegacyLogger):
    def write_log(self, message: str) -> None:
        self.log_legacy(message)

adapter = LoggerAdapter()
adapter.write_log("어댑터 패턴")

객체适格 예시

from abc import ABC, abstractmethod

class NotificationSender(ABC):
    @abstractmethod
    def dispatch(self, recipient: str, body: str) -> bool:
        pass

class MailProvider:
    def send_email(self, to: str, subject: str, content: str) -> dict:
        return {"code": 200, "message": "sent"}

class MailAdapter(NotificationSender):
    def __init__(self, provider: MailProvider, subject: str = "안내"):
        self.provider = provider
        self.subject = subject

    def dispatch(self, recipient: str, body: str) -> bool:
        response = self.provider.send_email(recipient, self.subject, body)
        return response.get("code") == 200

외부 API 연동: 통합 메시지 발송 시스템

이메일, SMS, 푸시 등 다양한 메시지 널을 하나의 인터페이스로 통합하는 예제입니다. 각 채널의 SDK는 서로 다른 메서드명과 반환 형식을 가집니다.

from abc import ABC, abstractmethod
from typing import Dict

# 목표 인터페이스
class MessageGateway(ABC):
    @abstractmethod
    def send(self, to: str, text: str) -> Dict[str, any]:
        pass

# 벤더 A: 이메일
class EmailVendor:
    def deliver(self, address: str, title: str, content: str):
        print(f"Email to {address}: {title}")
        return {"status": "ok", "trace_id": "E123"}

class EmailGateway(MessageGateway):
    def __init__(self, vendor: EmailVendor, title: str = "시스템 알림"):
        self.vendor = vendor
        self.title = title

    def send(self, to: str, text: str) -> Dict[str, any]:
        result = self.vendor.deliver(to, self.title, text)
        return {
            "success": result.get("status") == "ok",
            "trace": result.get("trace_id")
        }

# 벤더 B: SMS
class SmsVendor:
    def transmit(self, phone: str, msg: str):
        print(f"SMS to {phone}")
        return {"result_code": 0, "msg_id": "S456"}

class SmsGateway(MessageGateway):
    def __init__(self, vendor: SmsVendor):
        self.vendor = vendor

    def send(self, to: str, text: str) -> Dict[str, any]:
        result = self.vendor.transmit(to, text)
        return {
            "success": result.get("result_code") == 0,
            "trace": result.get("msg_id")
        }

# 클라이언트
def notify(gateway: MessageGateway, user: str, content: str):
    outcome = gateway.send(user, content)
    print("전송 성공" if outcome["success"] else "전송 실패")

이터 포맷 변환适格

최신 서비스는 JSON을 사용하지만 레거시 시스템은 XML을 요구하는 경우가 많습니다. 이 경우 변환 어댑터를 두어 비즈니스 로직에 영향을 주지 않고 포맷을 전환할 수 있습니다.

import json
import xml.etree.ElementTree as ET
from abc import ABC, abstractmethod
from typing import Dict

class DataFormatter(ABC):
    @abstractmethod
    def encode(self, payload: Dict) -> str:
        pass

    @abstractmethod
    def decode(self, raw: str) -> Dict:
        pass

class JsonFormatter(DataFormatter):
    def encode(self, payload: Dict) -> str:
        return json.dumps(payload, ensure_ascii=False)

    def decode(self, raw: str) -> Dict:
        return json.loads(raw)

class XmlEngine:
    def to_xml(self, data: Dict) -> str:
        root = ET.Element("payload")
        for key, value in data.items():
            child = ET.SubElement(root, key)
            child.text = str(value)
        return ET.tostring(root, encoding="unicode")

    def from_xml(self, raw: str) -> Dict:
        root = ET.fromstring(raw)
        return {child.tag: child.text for child in root}

class XmlFormatterAdapter(DataFormatter):
    def __init__(self, engine: XmlEngine):
        self.engine = engine

    def encode(self, payload: Dict) -> str:
        return self.engine.to_xml(payload)

    def decode(self, raw: str) -> Dict:
        return self.engine.from_xml(raw)

# 사용
json_formatter = JsonFormatter()
xml_formatter = XmlFormatterAdapter(XmlEngine())

data = {"user": "kim", "level": 5}
print(xml_formatter.encode(data))

FastAPI 기반 결제 게이트웨이 설계

실무에서는 FastAPI와 함께 어댑터 패턴을 활용해 결제, 알림, 스토리지 등 부 연동 모듈을 추상화합니다. 여기서는 결제 게이트웨이를 예시로 설명합니다.

프로젝트 구조

payment_gateway/
├── main.py
├── adapters/
│   ├── base.py
│   ├── kakao_pay.py
│   └── toss_pay.py
├── services/
│   └── payment_service.py
├── schemas/
│   └── payment.py

어댑터 기반 인터이스

from abc import ABC, abstractmethod
from decimal import Decimal
from enum import Enum
from typing import Dict
from datetime import datetime

class PaymentProvider(str, Enum):
    KAKAO = "kakao"
    TOSS = "toss"

class TransactionStatus(str, Enum):
    PENDING = "pending"
    COMPLETED = "completed"
    CANCELED = "canceled"

class PaymentAdapter(ABC):
    @property
    @abstractmethod
    def provider(self) -> PaymentProvider:
        pass

    @abstractmethod
    async def charge(self, order_id: str, amount: Decimal, item_name: str) -> Dict:
        pass

    @abstractmethod
    async def status(self, transaction_id: str) -> Dict:
        pass

    @abstractmethod
    async def cancel(self, transaction_id: str, amount: Decimal, reason: str) -> Dict:
        pass

카카오페이 어댑터 구현

import asyncio
from decimal import Decimal
from datetime import datetime

class KakaoPayAdapter(PaymentAdapter):
    @property
    def provider(self) -> PaymentProvider:
        return PaymentProvider.KAKAO

    async def charge(self, order_id: str, amount: Decimal, item_name: str) -> Dict:
        await asyncio.sleep(0.05)
        return {
            "success": True,
            "transaction_id": f"KP_{order_id}",
            "redirect_url": f"https://kakaopay.com/checkout/{order_id}"
        }

    async def status(self, transaction_id: str) -> Dict:
        await asyncio.sleep(0.05)
        return {
            "status": TransactionStatus.COMPLETED,
            "amount": Decimal("10000"),
            "settled_at": datetime.now()
        }

    async def cancel(self, transaction_id: str, amount: Decimal, reason: str) -> Dict:
        await asyncio.sleep(0.05)
        return {
            "success": True,
            "refund_id": f"REF_{transaction_id}",
            "message": "환불 완료"
        }

서비스 레이어에서 어댑터 등록

from fastapi import HTTPException

class PaymentService:
    def __init__(self):
        self._adapters: Dict[PaymentProvider, PaymentAdapter] = {
            PaymentProvider.KAKAO: KakaoPayAdapter(),
            PaymentProvider.TOSS: TossPayAdapter(),
        }

    def resolve(self, provider: PaymentProvider) -> PaymentAdapter:
        adapter = self._adapters.get(provider)
        if adapter is None:
            raise HTTPException(status_code=400, detail="지원하지 않는 결제 수단")
        return adapter

    async def create_order(self, provider: PaymentProvider, order_id: str, amount: Decimal, item_name: str):
        adapter = self.resolve(provider)
        return await adapter.charge(order_id, amount, item_name)

API 엔드포인트

from fastapi import FastAPI
from decimal import Decimal
from services.payment_service import PaymentService
from adapters.base import PaymentProvider

app = FastAPI(title="결제 게이트웨이")

service = PaymentService()

@app.post("/payments/{provider}")
async def create_payment(provider: PaymentProvider, order_id: str, amount: Decimal, item_name: str):
    return await service.create_order(provider, order_id, amount, item_name)

어댑터와 유사 패턴 비교

코드로 비교

# Adapter: 다른 인터페이스를 표준 인터페이스로 변경
class OldStorage:
    def put_file(self, path: str, data: bytes): ...

class StorageAdapter(Storage):
    def __init__(self, old: OldStorage):
        self.old = old

    def upload(self, path: str, data: bytes):
        self.old.put_file(path, data)

# Decorator: 동일 인터페이스에 기능 추가
class LoggingStorage(Storage):
    def __init__(self, storage: Storage):
        self.storage = storage

    def upload(self, path: str, data: bytes):
        print(f"업로드 시작: {path}")
        self.storage.upload(path, data)

# Proxy: 동일 인터페이스에 접근 제어
class SecureStorage(Storage):
    def __init__(self, storage: Storage):
        self.storage = storage

    def upload(self, path: str, data: bytes):
        if not self.is_authorized():
            raise PermissionError()
        self.storage.upload(path, data)

설계 및 구현 팁

• 어댑터 내부에는 비즈니스 로직을 두지 말고 인터페이스 변환에만 집중합니다.
• 단위 테스트 시 실제 외부 서비스 대신 가짜 어댑터를 주입할 수 있도록 설계합니다.
• 예외 처리를 어댑터 레벨에서 일관되게 매핑하여 상위 계층에서 벤더별 세부 정보를 몰라도 되게 합니다.
• 새로운 벤더 추가 시 기존 클라이언트 코드를 수정하지 않도록 의존성 역전 원칙을 수합니다.
• 비동기 환경에서는 어댑터 인터페이스도 async 메서드로 설계하여 이벤트 루프를 효율적으로 활용합니다.