이 글은 누구를 위한 것인가
- 재고 없이 이커머스를 운영하고 싶은 팀
- 여러 공급사와 연동해서 자동으로 주문을 넘기고 싶은 개발자
- 드롭쉬핑 마진 관리와 가격 동기화를 자동화하려는 팀
들어가며
드롭쉬핑의 핵심은 공급사 재고와 우리 쇼핑몰 재고를 실시간으로 동기화하고, 주문이 들어오면 자동으로 공급사에 전달하는 것이다. 수동으로 하면 품절 주문과 배송 지연이 생긴다. 공급사 API를 호출할 때는 재시도로 중복 주문이 나가지 않도록 멱등 요청 패턴을 적용해야 한다.
이 글은 bluefoxdev.kr의 드롭쉬핑 자동화 가이드 를 참고하여 작성했습니다.
1. 드롭쉬핑 아키텍처
[드롭쉬핑 자동화 흐름]
재고 동기화:
공급사 API (1시간마다 폴링)
↓
재고 변경 감지
↓
우리 쇼핑몰 재고 업데이트
↓
가격 재계산 (원가 + 마진율)
주문 처리:
고객 주문 → 결제 완료
↓
공급사 자동 발주 (API)
↓
공급사 송장번호 수신
↓
고객에게 배송 추적 정보 전달
[공급사 API 유형]
REST API: 실시간 조회, 주문 전달 가능
FTP/SFTP: 일괄 재고 파일 교환
이메일 자동화: API 없는 공급사 대안
EDI: 대형 공급사 표준
2. 공급사 연동 시스템
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class SupplierProduct:
supplier_id: str
supplier_sku: str
name: str
cost_price: int
stock_quantity: int
lead_time_days: int
@dataclass
class SupplierOrder:
our_order_id: str
supplier_order_id: str | None
status: str
tracking_number: str | None
class SupplierAdapter(ABC):
"""공급사별 어댑터 패턴"""
@abstractmethod
async def get_inventory(self) -> list[SupplierProduct]:
pass
@abstractmethod
async def place_order(self, order: dict) -> SupplierOrder:
pass
@abstractmethod
async def get_order_status(self, supplier_order_id: str) -> SupplierOrder:
pass
class RestApiSupplierAdapter(SupplierAdapter):
"""REST API 기반 공급사 어댑터"""
def __init__(self, base_url: str, api_key: str, supplier_id: str):
self.base_url = base_url
self.api_key = api_key
self.supplier_id = supplier_id
async def get_inventory(self) -> list[SupplierProduct]:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.base_url}/inventory",
headers={"X-API-Key": self.api_key},
)
return [
SupplierProduct(
supplier_id=self.supplier_id,
supplier_sku=item["sku"],
name=item["name"],
cost_price=int(item["price"]),
stock_quantity=item["stock"],
lead_time_days=item.get("lead_time", 3),
)
for item in response.json()["products"]
]
async def place_order(self, order: dict) -> SupplierOrder:
payload = {
"reference": order["our_order_id"],
"shipping_address": order["delivery_address"],
"items": [
{"sku": item["supplier_sku"], "quantity": item["quantity"]}
for item in order["items"]
],
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/orders",
json=payload,
headers={"X-API-Key": self.api_key},
)
data = response.json()
return SupplierOrder(
our_order_id=order["our_order_id"],
supplier_order_id=data["order_id"],
status="placed",
tracking_number=None,
)
async def sync_supplier_inventory(supplier_id: str):
"""공급사 재고 동기화"""
adapter = get_supplier_adapter(supplier_id)
products = await adapter.get_inventory()
for product in products:
# 기존 매핑 조회
mapping = await get_supplier_product_mapping(
supplier_id, product.supplier_sku
)
if not mapping:
continue
our_product_id = mapping["our_product_id"]
# 재고 업데이트
await update_product_stock(our_product_id, product.stock_quantity)
# 가격 업데이트 (마진 + 배송비 포함)
margin_rate = mapping.get("margin_rate", 0.30)
shipping_cost = mapping.get("shipping_cost", 3000)
our_price = round(product.cost_price * (1 + margin_rate) + shipping_cost)
await update_product_price(our_product_id, our_price)
await log_sync_result(supplier_id, len(products))
async def auto_dropship_order(order_id: str):
"""주문 자동 발주"""
order = await get_order_with_items(order_id)
# 공급사별로 아이템 그룹화
supplier_groups = {}
for item in order["items"]:
mapping = await get_product_supplier_mapping(item["product_id"])
if not mapping:
continue
sid = mapping["supplier_id"]
if sid not in supplier_groups:
supplier_groups[sid] = []
supplier_groups[sid].append({
**item,
"supplier_sku": mapping["supplier_sku"],
})
# 각 공급사에 발주
for supplier_id, items in supplier_groups.items():
adapter = get_supplier_adapter(supplier_id)
supplier_order = await adapter.place_order({
"our_order_id": order_id,
"delivery_address": order["delivery_address"],
"items": items,
})
await save_supplier_order(order_id, supplier_id, supplier_order)
마무리
드롭쉬핑 자동화에서 가장 중요한 것은 재고 동기화 주기다. 공급사 재고는 빠르게 변하므로, 최소 1시간마다 동기화해야 "재고 있음"으로 표시된 상품이 품절되는 사태를 막을 수 있다. 재고 10개 이하는 실시간 확인 로직을 추가하는 것이 안전하다.