API로 사내외 서비스를 연결하기

API가 필요한 이유

엑셀과 데이터베이스 자동화는 회사 안의 데이터를 정리하고 저장하는 데 강하다. 하지만 실제 업무는 한 시스템 안에서 끝나지 않는다. 고객 정보는 CRM에 있고, 주문과 재고는 ERP에 있으며, 결제 결과는 PG사 서비스에 있고, 담당자 알림은 슬랙이나 팀즈 같은 협업 도구로 전달된다.

이때 API는 시스템 사이의 통로가 된다. 파이썬 스크립트는 API를 호출해 고객 정보를 가져오고, 주문 상태를 바꾸고, 처리 결과를 협업 도구에 알릴 수 있다. 즉 API를 이해하면 파이썬은 단순한 파일 처리 도구가 아니라 여러 비즈니스 시스템을 연결하는 허브가 된다.

API 자동화의 기본 흐름은 앞 장에서 다룬 데이터 처리 흐름과 같다.

1. 필요한 데이터를 외부 시스템에서 가져온다.

2. 파이썬에서 검증하고 업무 규칙을 적용한다.

3. 다른 시스템에 결과를 반영한다.

4. 처리 결과를 데이터베이스와 로그에 남긴다.

5. 담당자에게 알린다.

AI는 이 흐름 중 일부에 들어갈 수 있다. 예를 들어 고객 문의 내용을 분류하거나 주문 메모를 요약할 수 있다. 그러나 고객 등급 조회, 주문 상태 변경, 알림 발송처럼 규칙과 대상 시스템이 명확한 작업은 API 기반 자동화가 더 안정적이다.

AI 생성 이미지

REST API를 업무 언어로 이해하기

실무에서 가장 자주 만나는 방식은 REST API다. REST API는 고객, 주문, 결제, 메시지 같은 업무 대상을 URL로 표현하고, HTTP 메서드로 어떤 작업을 할지 지정한다. 대표 메서드는 GET, POST, PUT, PATCH, DELETE이며 각각 조회, 생성, 전체 수정, 일부 수정, 삭제에 대응하는 경우가 많다.¹

API 호출은 요청과 응답으로 나뉜다. 요청에는 주소, 메서드, 인증 정보, 파라미터, 본문 데이터가 들어간다. 응답에는 상태 코드, 응답 본문, 헤더가 들어간다. 본문 데이터는 대부분 JSON 형식이다. JSON은 파이썬의 딕셔너리와 비슷한 구조라 자동화 코드에서 다루기 쉽다.

상태 코드는 API 호출 결과를 짧은 숫자로 알려준다. 200은 조회 성공, 201은 생성 성공, 204는 성공했지만 본문이 없다는 뜻이다. 반대로 400은 요청 형식 오류, 401은 인증 실패, 404는 대상 없음, 422는 값 검증 실패, 429는 호출 제한 초과, 500번대는 서버 오류를 의미한다. 서비스마다 세부 의미는 조금씩 다를 수 있으므로, 실제 자동화에서는 해당 API 명세서의 오류 설명을 함께 확인해야 한다.

초보자가 자주 하는 실수는 응답 본문만 보고 성공 여부를 판단하는 것이다. 운영용 자동화에서는 반드시 상태 코드를 확인해야 한다. 특히 주문 상태 변경, 결제 취소, 고객 정보 수정처럼 결과가 업무에 직접 영향을 주는 API는 실패했는데도 성공한 것처럼 넘어가면 안 된다.

인증과 비밀 정보 관리

대부분의 업무 API는 인증이 필요하다. 인증은 이 요청을 보낸 사람이 누구인지 확인하는 절차다. 흔한 방식은 세 가지다.

• API Key: 서비스에서 발급한 키를 헤더나 파라미터에 넣는다.

• Bearer Token: 로그인이나 발급 절차로 받은 토큰을 Authorization 헤더에 넣는다.

• OAuth: 사용자가 특정 권한을 승인하면 앱이 제한된 범위에서 API를 호출한다.

중요한 원칙은 간단하다. API 키와 토큰을 코드에 직접 쓰지 않는다. 앞 장에서 다룬 것처럼 .env 파일이나 환경 변수에 저장하고, 저장소에는 올리지 않는다. 로그에도 토큰이 찍히지 않게 해야 한다. 또한 자동화 계정에는 필요한 권한만 부여한다. 고객 조회만 필요한 스크립트에 고객 삭제 권한까지 주는 것은 위험하다.

여기서 한 가지를 더 조심해야 한다. CRM 토큰, ERP 토큰, 협업 도구 웹훅은 서로 다른 서비스의 비밀 정보다. 한 서비스의 토큰을 다른 서비스 요청에 함께 보내지 않도록 헤더를 분리해야 한다. 특히 여러 API를 하나의 파이썬 코드에서 연결할 때는 "편하니까 모든 요청에 같은 인증 헤더를 붙인다"는 방식이 보안 사고로 이어질 수 있다.

requests로 API 호출 구조 만들기

파이썬에서는 requests 라이브러리를 사용하면 API 요청을 간단히 보낼 수 있다. requests.get(), requests.post(), requests.patch()처럼 메서드별 함수를 사용하고, JSON 본문은 json= 인자로 전달하는 방식이 일반적이다.²

아래 코드는 여러 API 호출에서 공통으로 쓸 수 있는 기본 함수다. 핵심은 타임아웃, 상태 코드 확인, 재시도, JSON 파싱을 한곳에 모으는 것이다. 실제 서비스에서는 엔드포인트와 필드 이름이 다르므로, 구조를 참고해 각 회사의 API 명세에 맞게 바꿔 쓰면 된다.

import os
import time
import logging
import requests
from dotenv import load_dotenv

load_dotenv()

CRM_BASE_URL = os.getenv('CRM_BASE_URL')
ERP_BASE_URL = os.getenv('ERP_BASE_URL')
CHAT_WEBHOOK_URL = os.getenv('CHAT_WEBHOOK_URL')
API_TOKEN = os.getenv('BUSINESS_API_TOKEN')

required_env = {
    'CRM_BASE_URL': CRM_BASE_URL,
    'ERP_BASE_URL': ERP_BASE_URL,
    'CHAT_WEBHOOK_URL': CHAT_WEBHOOK_URL,
    'BUSINESS_API_TOKEN': API_TOKEN,
}
missing = [name for name, value in required_env.items() if not value]
if missing:
    raise RuntimeError(f'missing environment variables: {", ".join(missing)}')

os.makedirs('logs', exist_ok=True)
logging.basicConfig(
    filename='logs/api_automation.log',
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s'
)

session = requests.Session()


def build_headers(auth=True, extra_headers=None):
    headers = {
        'Accept': 'application/json'
    }

    if auth:
        headers['Authorization'] = f'Bearer {API_TOKEN}'

    if extra_headers:
        headers.update(extra_headers)

    return headers


def request_api(method, url, *, auth=True, **kwargs):
    extra_headers = kwargs.pop('headers', None)
    headers = build_headers(auth=auth, extra_headers=extra_headers)

    if 'json' in kwargs:
        headers['Content-Type'] = 'application/json'

    for attempt in range(3):
        try:
            response = session.request(
                method,
                url,
                headers=headers,
                timeout=10,
                **kwargs
            )

            if response.status_code in [429, 500, 502, 503, 504]:
                retry_after = response.headers.get('Retry-After')
                if retry_after and retry_after.isdigit():
                    wait_seconds = int(retry_after)
                else:
                    wait_seconds = 2 ** attempt

                logging.warning(
                    f'retry {method} {url} after {wait_seconds}s '
                    f'status={response.status_code}'
                )
                time.sleep(wait_seconds)
                continue

            response.raise_for_status()

            if response.status_code == 204:
                return None

            content_type = response.headers.get('Content-Type', '')
            if 'application/json' in content_type:
                return response.json()
            return response.text

        except requests.Timeout as error:
            logging.warning(f'timeout {method} {url} attempt={attempt + 1}')
            if attempt == 2:
                raise RuntimeError(f'api timeout after retries: {method} {url}') from error
            time.sleep(2 ** attempt)
        except requests.HTTPError as error:
            logging.error(f'http error {method} {url} {error}')
            raise
        except requests.RequestException as error:
            logging.error(f'request error {method} {url} {error}')
            raise

    raise RuntimeError(f'api failed after retries: {method} {url}')

이 함수는 모든 문제를 해결하지는 않는다. 하지만 운영에 필요한 기본 습관을 담고 있다. 외부 API는 언제든 느려질 수 있으므로 타임아웃을 둔다. 일시적 오류는 짧게 기다린 뒤 재시도한다. 429 응답에 Retry-After 헤더가 있으면 서비스가 알려준 대기 시간을 우선 사용한다. 실패는 로그에 남기고, 잘못된 응답을 조용히 무시하지 않는다.

또 하나 중요한 점은 auth=True 옵션이다. CRM이나 ERP처럼 업무 API 토큰이 필요한 요청에는 인증 헤더를 붙이고, 협업 도구 웹훅처럼 별도 URL 자체가 인증 역할을 하는 요청에는 인증 헤더를 붙이지 않을 수 있다. 서비스마다 인증 방식이 다르므로, 공통 함수 안에서도 인증 정보를 무조건 한꺼번에 붙이지 않는 편이 안전하다.

고객 조회, 주문 변경, 알림 발송 연결하기

이제 업무 흐름을 코드로 연결해 보자. 예시는 고객 이메일로 CRM에서 고객을 찾고, 조건에 따라 ERP 주문 상태를 변경한 뒤, 협업 도구로 결과를 알리는 흐름이다.

def find_customer_by_email(email):
    url = f'{CRM_BASE_URL}/customers'
    data = request_api('GET', url, params={'email': email})
    items = data.get('items', [])
    return items[0] if items else None


def update_order_status(order_id, status):
    url = f'{ERP_BASE_URL}/orders/{order_id}'
    payload = {'status': status}
    return request_api('PATCH', url, json=payload)


def send_notification(message):
    payload = {'text': message}
    return request_api('POST', CHAT_WEBHOOK_URL, auth=False, json=payload)


def process_order(order_id, customer_email):
    customer = find_customer_by_email(customer_email)

    if customer is None:
        send_notification(f'고객을 찾지 못했습니다. order_id={order_id}')
        return

    if customer.get('grade') == 'VIP':
        update_order_status(order_id, 'priority_shipping')
        send_notification(f'VIP 주문을 우선 배송으로 변경했습니다. order_id={order_id}')
    else:
        send_notification(f'일반 주문 확인 완료. order_id={order_id}')

실제 회사에서는 이 코드 앞뒤에 더 많은 검증이 붙는다. 주문 ID가 이미 처리된 적 있는지 데이터베이스에서 확인해야 한다. 변경 전 상태가 수정 가능한 상태인지도 봐야 한다. VIP 기준이 바뀔 수 있다면 코드에 직접 박아두지 말고 설정 파일이나 기준 테이블로 관리하는 편이 좋다.

특히 PATCH나 POST처럼 데이터를 바꾸는 요청은 중복 실행에 주의해야 한다. 네트워크 오류로 응답을 받지 못했지만 실제 서버에서는 변경이 끝났을 수 있다. 이런 경우 같은 요청을 다시 보내면 중복 처리될 수 있다. 결제, 환불, 포인트 지급 같은 업무는 요청 고유 번호를 저장하고, 가능하면 API가 제공하는 멱등성 키(idempotency key)를 사용하며, 필요하면 사람의 승인 단계를 둔다.

페이지네이션과 호출 제한 처리하기

API는 데이터를 한 번에 모두 주지 않는 경우가 많다. 고객 10만 명을 한 번에 내려주면 서버와 클라이언트 모두 부담이 크기 때문이다. 그래서 페이지네이션을 사용한다. 페이지네이션은 데이터를 여러 묶음으로 나누어 가져오는 방식이다.

def fetch_updated_customers(updated_after):
    page = 1
    all_items = []

    while True:
        url = f'{CRM_BASE_URL}/customers'
        data = request_api(
            'GET',
            url,
            params={
                'updated_after': updated_after,
                'page': page,
                'limit': 100
            }
        )

        items = data.get('items', [])
        all_items.extend(items)

        if not data.get('has_more'):
            break

        page += 1

    return all_items

API에 따라 방식은 다를 수 있다. 어떤 서비스는 page와 limit를 쓰고, 어떤 서비스는 next_cursor를 응답으로 내려준다. 중요한 것은 명세서에서 다음 데이터를 가져오는 방법을 확인하고, 마지막 페이지 조건을 정확히 구현하는 것이다. 데이터가 매우 많다면 모든 항목을 한 번에 리스트에 담기보다 페이지 단위로 처리하고 저장하는 방식도 고려해야 한다.

호출 제한도 반드시 봐야 한다. 많은 서비스는 1분에 60회, 하루에 1만 회처럼 호출 횟수를 제한한다. 제한을 넘으면 429 상태 코드가 돌아온다. 이때 무작정 반복 호출하면 더 오래 차단될 수 있다. 응답 헤더에 남은 호출 수나 재시도 가능 시간이 제공된다면 그 값을 기준으로 기다리는 것이 좋다.

재시도는 모든 오류에 적용하지 않는다. 500번대 오류나 429처럼 일시적일 가능성이 큰 오류는 재시도할 수 있다. 하지만 400, 401, 404, 422는 요청 자체가 잘못된 경우가 많다. 이런 오류는 데이터를 고치거나 인증 설정을 확인해야 한다.

API 명세서에서 필요한 것만 찾기

API 명세서는 처음 보면 복잡하다. 하지만 자동화 관점에서는 필요한 항목이 정해져 있다.

명세서를 읽을 때는 전체를 외우려 하지 말고 업무 질문에서 출발한다. 예를 들어 목표가 미결제 주문을 찾아 배송 보류로 바꾸는 것이라면 다음 세 가지만 먼저 찾는다.

1. 미결제 주문 목록을 조회하는 엔드포인트

2. 응답에서 필요한 주문 ID와 결제 상태 필드

3. 주문 상태를 배송 보류로 변경하는 엔드포인트

그다음 운영 조건을 확인한다. 한 번에 몇 건까지 조회할 수 있는지, 호출 제한은 얼마인지, 이미 배송이 시작된 주문도 변경 가능한지, 실패했을 때 어떤 오류 코드가 오는지 살핀다. 이 정보가 있어야 안전한 자동화가 된다.

안정적으로 운영하기 위한 점검 목록

API 자동화는 연결이 쉬운 만큼 장애 지점도 많다. 외부 서비스가 느려질 수 있고, 토큰이 만료될 수 있으며, 명세가 바뀔 수도 있다. 그래서 운영 전 다음 항목을 점검해야 한다.

• 입력 데이터 검증: 고객 이메일, 주문 ID, 날짜 형식이 올바른가?

• 중복 실행 방지: 이미 처리한 주문을 다시 처리하지 않는가?

• 로그 기록: 요청 대상, 처리 결과, 오류 원인을 추적할 수 있는가?

• 비밀 정보 보호: 토큰이 코드와 로그에 노출되지 않는가?

• 인증 정보 분리: 서로 다른 서비스에 같은 토큰을 잘못 보내지 않는가?

• 권한 최소화: 자동화 계정이 필요한 권한만 갖고 있는가?

• 호출 제한 대응: 제한을 넘었을 때 기다리거나 다음 배치로 넘길 수 있는가?

• 실패 알림: 반복 실패 시 담당자가 알 수 있는가?

• 승인 절차: 결제 취소, 환불, 대량 수정은 사람이 확인하는가?

API는 자동화의 연결부다. 파이썬은 CRM, ERP, 결제, 협업 도구를 이어 붙이고, 데이터베이스는 처리 이력을 남기며, 사람은 중요한 예외와 승인 판단을 맡는다. 이 구조가 잡히면 이후 이메일, 문서, 협업 도구 자동화도 같은 원리로 확장할 수 있다.

참고

¹Python and REST APIs: Interacting With Web Services

²Automating REST API Interactions Using Python: A Comprehensive Guide