Python에서 회사 이메일을 찾으려면 이름과 회사 정보 또는 공개 LinkedIn 프로필 URL을 Requests 라이브러리로 Datablist의 Waterfall Email Finder API에 전송하면 됩니다. API는 찾은 회사 이메일과 해당 주소의 전송 가능 상태, 메일 제공업체를 반환합니다.
아래의 전체 예제에서는 microsoft.com의 Satya Nadella와 greylock.com의 Reid Hoffman을 조회합니다. 백그라운드 작업이 끝날 때까지 기다린 다음 모든 결과 페이지를 가져옵니다. 이후 이 결과를 재사용해 캠페인 검토용으로 인증된 연락처의 로컬 목록을 만들겠습니다.
📌 바쁜 분을 위한 요약
하나의 Python 함수로 알려진 연락처를 Waterfall Email Finder API에 전송하고, 작업이 끝날 때까지 기다린 뒤 찾은 회사 이메일을 수집하세요. 미확인 연락처는 원본 목록에 그대로 보관하면서 인증된 연락처만 캠페인 검토용으로 준비할 수 있습니다.
- Input: 이름과 성 및 회사 도메인, 전체 이름과 회사명 또는 공개 LinkedIn 프로필 URL
- Output:
emailStatus와mxProvider가 포함된 회사 이메일 및 이메일이 있는 연락처만 담은 로컬 검토 목록
💡 No-code 방식이 더 편한가요? 코드를 작성하지 않고 컬렉션에서 회사 이메일을 찾으려면 Datablist의 Waterfall Email Finder를 사용하세요.
Python, Requests 및 API 키 설정
이 Python 예제에서는 requests 패키지를 사용합니다. find_work_emails.py라는 파일을 만든 후 Requests를 설치하세요.
python -m pip install requestsDatablist API 계정 페이지에서 Personal API Key를 생성하세요. REST API를 이용하려면 Growth 플랜이 필요합니다. 스크립트를 실행할 터미널에서 키를 환경 변수로 내보냅니다.
export DATABLIST_API_KEY="your-api-key"Waterfall Email Finder가 이메일을 찾는 데 성공하면 크레딧도 사용됩니다. 아래 비용 섹션에서는 이 크레딧과 API 이용 권한을 제공하는 구독료를 구분해 설명합니다.
하나의 Python 함수로 회사 이메일 찾기
아래의 전체 스크립트를 find_work_emails.py에 붙여 넣으세요.
import json
import os
import time
import uuid
import requests
def find_emails(people, api_key, settings):
"""Submit one email-finder task, wait, and return every result."""
tasks_url = "https://api.datablist.com/v1/enrichments/tasks"
headers = {"X-API-Key": api_key}
unique_id = str(uuid.uuid4())
print(f"Task uniqueId: {unique_id}", flush=True)
payload = {
"enrichmentId": "waterfall-email-finder",
"uniqueId": unique_id,
"settings": settings,
"inputs": people,
}
response = requests.post(tasks_url, headers=headers, json=payload)
response.raise_for_status()
task = response.json()
task_id = task["@id"]
task_url = f"{tasks_url}/{task_id}"
while task["status"] not in ("completed", "failed", "stopped"):
time.sleep(3)
response = requests.get(task_url, headers=headers)
response.raise_for_status()
task = response.json()
if task["status"] != "completed":
raise RuntimeError(
f"Task {task_id} {task['status']}: {task.get('error')}"
)
results = []
results_url = f"{task_url}/results"
params = {"limit": 100}
while True:
response = requests.get(results_url, headers=headers, params=params)
response.raise_for_status()
page = response.json()
results.extend(page["results"])
next_cursor = page["nextCursor"]
if next_cursor is None:
break
params["cursor"] = next_cursor
return results
if __name__ == "__main__":
api_key = os.environ.get("DATABLIST_API_KEY")
if not api_key:
raise RuntimeError("Set DATABLIST_API_KEY before running this script.")
settings = {"sourceKind": "name_company"}
people = [
{
"inputId": "contact-satya-nadella",
"data": {
"firstName": "Satya",
"lastName": "Nadella",
"companyNameOrDomain": "microsoft.com",
},
},
{
"inputId": "contact-reid-hoffman",
"data": {
"firstName": "Reid",
"lastName": "Hoffman",
"companyNameOrDomain": "greylock.com",
},
},
]
lookup_results = find_emails(people, api_key, settings)
for result in lookup_results:
print(json.dumps(result, indent=2))같은 터미널에서 실행하세요.
python find_work_emails.py지속적으로 관리되는 Requests 예제도 다운로드할 수 있습니다. Blocking 방식이 이해되었다면 저장소에서 비동기 HTTPX 예제도 확인해 보세요.
함수는 순서대로 세 가지 API 호출을 수행합니다.
첫째, 작업을 제출합니다. 각 inputId는 연락처를 식별하는 고정 ID입니다. 중첩된 data 객체에는 조회에 사용할 필드가 들어갑니다. POST 요청은 이후 호출에 쓰이는 작업 ID인 @id와 함께 202 Accepted를 반환합니다. 응답이 유실되더라도 복구에 사용할 값을 확보할 수 있도록 코드는 제출 전에 uniqueId를 출력합니다.
둘째, 작업 상태를 폴링합니다. Datablist는 작업을 백그라운드에서 처리합니다. 이 Requests 클라이언트는 작업 상태가 completed, failed 또는 stopped가 될 때까지 3초마다 확인합니다. 기다리는 동안 Requests 호출은 blocking 상태지만 API 작업 자체는 비동기로 실행됩니다.
셋째, 결과를 가져옵니다. 하나의 작업 결과가 여러 페이지로 나뉠 수 있습니다. 루프는 nextCursor가 null이 될 때까지 페이지를 계속 가져옵니다. 응답 순서가 아니라 inputId를 사용해 각 결과를 원본 연락처와 연결하세요.
연락처 데이터에 맞는 입력 방식 선택
하나의 작업에서는 하나의 입력 모드만 사용할 수 있습니다. 동일한 find_emails() 함수를 유지하고, 원본 데이터에 맞게 settings와 people 값만 교체하세요.
가능하면 작업 내에서 고유하면서 원본 연락처에 이미 할당된 값을 inputId로 선택하세요. CRM 레코드 ID나 변경되지 않는 Lead ID가 적합합니다. Datablist는 결과와 함께 이 값을 반환하므로 이름을 비교하거나 응답이 입력 순서대로 올 것이라 가정하지 않고도 레코드를 매칭할 수 있습니다.
이름과 성, 회사 도메인이 각각 있다면 sourceKind: "name_company"를 사용하세요. 확인된 도메인이 가장 명확한 회사 식별자입니다.
settings = {"sourceKind": "name_company"}
people = [
{
"inputId": "contact-satya-nadella",
"data": {
"firstName": "Satya",
"lastName": "Nadella",
"companyNameOrDomain": "microsoft.com",
},
}
]
lookup_results = find_emails(people, api_key, settings)원본 데이터에 이름 필드가 하나만 있고 회사명이 있다면 useFullName을 활성화하세요.
settings = {"sourceKind": "name_company", "useFullName": True}
people = [
{
"inputId": "contact-satya-nadella",
"data": {
"fullName": "Satya Nadella",
"companyNameOrDomain": "Microsoft",
},
}
]
lookup_results = find_emails(people, api_key, settings)공개 개인 프로필 URL이 있다면 LinkedIn 방식을 사용하세요.
settings = {"sourceKind": "linkedin"}
people = [
{
"inputId": "contact-satya-nadella",
"data": {
"linkedInProfile": "https://www.linkedin.com/in/satyanadella/"
},
}
]
lookup_results = find_emails(people, api_key, settings)이 방식은 이메일 조회를 위한 인물 정보로 공개 linkedin.com/in/... 프로필 URL을 받습니다. Sales Navigator 프로필 URL은 지원하지 않습니다.
Provider 키와 적용 순서를 직접 선택해야 한다면 Custom Waterfall 가이드를 참고하세요. 여기서 사용하는 Managed Waterfall은 Provider별 설정이 필요하지 않습니다.
Managed Waterfall의 이메일 탐색 원리
이름과 회사 정보를 사용하는 요청에서 API는 제공된 도메인을 바로 사용합니다. 회사명을 제공하면 먼저 해당 이름을 사용할 수 있는 회사 도메인으로 변환합니다. LinkedIn 요청에서는 공개 프로필 URL을 정리한 뒤, 확보된 인물 및 회사 정보를 조건에 맞는 이메일 Provider가 사용할 수 있는 컨텍스트로 변환합니다.
Managed Waterfall은 아직 이메일을 찾지 못한 연락처를 조건에 맞는 Provider로 보냅니다. 사용할 수 있는 이메일이 반환되지 않으면 다른 Provider가 같은 연락처를 다시 조회할 수 있습니다. 회사 이메일이 승인되는 즉시 해당 연락처의 조회 과정은 중단됩니다. 기본 응답에는 이메일을 찾은 Provider의 이름이 표시되지 않습니다.
글로벌 Outbound에 Waterfall이 효과적인 이유
회사 이메일 Provider마다 사용하는 데이터셋이 달라 국가와 시장별 커버리지에도 차이가 있습니다. 한 Provider가 특정 지역의 연락처에서 결과를 찾지 못해도 다른 Provider에는 해당 이메일을 찾는 데 필요한 데이터가 있을 수 있습니다.
Sales 또는 Growth 팀은 Waterfall을 통해 미확인 연락처를 Outbound 목록에서 제외하기 전에 여러 번 탐색할 수 있습니다. 기술 프로세스는 일관되게 유지하면서 단일 Provider 방식으로 놓칠 수 있는 연락처를 추가로 확보할 수 있습니다.
Datablist는 하나의 API 연동 안에서 조건에 맞는 Provider의 실행 순서를 관리합니다. Provider마다 별도의 연동을 구축하고 응답을 직접 병합할 필요 없이, 연락처를 한 번 제출하고 통일된 형식의 결과를 받을 수 있습니다.
전체 API 결과 확인하기
예제 작업에서는 두 건 모두 성공 결과가 반환되었습니다. 아래 이메일 주소는 일부를 가렸습니다.
{
"taskId": "2286103ce43748828e27ce71b9cc526b",
"results": [
{
"inputId": "contact-satya-nadella",
"status": "success",
"outputs": {
"gender": null,
"mxProvider": "microsoft",
"emailFound": "s*******@microsoft.com",
"emailStatus": "verified"
},
"creditsUsed": "25",
"error": null
},
{
"inputId": "contact-reid-hoffman",
"status": "success",
"outputs": {
"gender": null,
"mxProvider": "mimecast",
"emailFound": "r*******@greylock.com",
"emailStatus": "verified"
},
"creditsUsed": "25",
"error": null
}
],
"nextCursor": null
}emailFound는 반환된 회사 이메일입니다. emailStatus는 조회 시점의 전송 가능 상태를 나타내며, mxProvider는 확인 가능한 경우 이메일 호스팅 Provider를 표시합니다. 두 레코드 모두 verified 상태이고 각각 25크레딧을 사용했습니다.
아래 캠페인 코드는 verified 결과를 campaign_contacts에 넣고, 그 밖의 상태로 확인된 이메일은 needs_verification에 보관합니다. 각 결과에 실제로 청구된 크레딧은 creditsUsed 값을 기준으로 확인해야 합니다.
기본 Managed Waterfall 결과에는 finderProvider가 포함되지 않습니다. 이 필드는 유효한 Custom Key 기반 이름·회사 조회 워크플로에서만 조건부로 출력되므로, 스크립트에서 mxProvider를 근거로 이를 추정해서는 안 됩니다.
찾은 이메일을 캠페인 검토용으로 준비
직접 조회가 정상적으로 작동하면 lookup_results가 반환된 뒤 스크립트 끝에 다음 블록을 추가하세요. 이미 실행한 조회 결과를 바탕으로 캠페인 검토용 로컬 연락처를 준비합니다.
results_by_id = {result["inputId"]: result for result in lookup_results}
campaign_contacts = []
needs_verification = []
for person in people:
result = results_by_id.get(person["inputId"])
if result is None:
continue
if result["status"] != "success":
continue
if not result["outputs"].get("emailFound"):
continue
contact = person.copy()
contact["result"] = result
if result["outputs"].get("emailStatus") == "verified":
campaign_contacts.append(contact)
else:
needs_verification.append(contact)
print(json.dumps({
"campaign_contacts": campaign_contacts,
"needs_verification": needs_verification,
}, indent=2))results_by_id를 사용하면 응답 순서는 중요하지 않습니다. Satya의 경우 루프가 원본 people 목록에서 contact-satya-nadella를 읽고, 동일한 inputId를 가진 결과를 찾은 뒤 원본 레코드를 복사해 전체 API 객체를 result 아래에 추가합니다. 이메일 주소가 있고 상태가 verified이므로 이 연락처는 campaign_contacts에 들어갑니다. Reid의 API 결과가 먼저 도착하더라도 같은 과정을 거칩니다.
세 가지 조건을 통해 판단 로직을 명확하게 확인할 수 있습니다. 첫 번째 조건은 일치하는 결과가 반환되지 않은 연락처를 건너뜁니다. 두 번째 조건은 행 단위 오류와 success가 아닌 상태를 제외합니다. 세 번째 조건은 실제 emailFound 값이 있는지 확인합니다. 전체 result 객체를 연결하면 상태, 출력값, 크레딧 사용량과 오류 필드를 나중에 검토할 수 있도록 보존할 수 있습니다.
다른 전송 가능 상태의 이메일은 needs_verification으로 이동합니다. 결과가 없거나, 행 오류가 있거나, emailFound가 없는 결과는 준비된 두 목록 어디에도 포함되지 않습니다. 원본 people 목록은 수정이나 추후 조사를 위해 그대로 유지됩니다.
Waterfall Email Finder API 비용 계산
기본 Managed Waterfall은 찾은 회사 이메일 한 건당 25크레딧을 사용합니다. 기본 조회에서 이메일을 찾지 못하면 크레딧을 사용하지 않습니다. 제출한 전체 연락처 수에 일정 값을 곱하지 말고 각 결과의 creditsUsed를 확인하세요.
unverified 상태의 이메일도 찾는 데 성공했다면 25크레딧을 사용합니다. 로컬 캠페인 코드에서 campaign_contacts가 아닌 needs_verification에 분류되더라도 동일합니다.
이 예제에서 찾은 이메일 두 건에는 50크레딧이 사용되었습니다. 비용을 계획할 때는 다음을 참고하세요.
- 회사 이메일 100건을 찾으면 2,500크레딧을 사용합니다.
- 회사 이메일 1,000건을 찾으면 25,000크레딧을 사용합니다.
- 결과가 없는 조회에는 비용이 청구되지 않으므로 실제 사용량은
연락처 수 × 25보다 적을 수 있습니다.
REST API를 이용하려면 Growth 플랜이 필요합니다. 현재 USD 추가 구매 요율은 20달러에 20,000크레딧으로, 크레딧당 0.001달러입니다. 이 요율을 적용하면 25,000크레딧의 가치는 25달러입니다. 구독료는 이 크레딧 계산과 별도입니다. 최신 플랜과 패키지 정보는 Datablist 요금 페이지에서 확인하세요.
자주 발생하는 오류 해결과 작업 복구
API 키가 없습니다. 스크립트는 요청을 보내기 전에 Set DATABLIST_API_KEY before running this script. 오류를 발생시킵니다. 현재 터미널에서 환경 변수를 내보낸 뒤 파일을 다시 실행하세요.
API가 인증 오류를 반환합니다. raise_for_status()가 HTTP 오류를 표시합니다. 현재 유효한 Personal API Key를 복사하고 X-API-Key 헤더가 해당 키를 사용하는지 확인한 후, 인증 정보를 수정한 뒤 다시 실행하세요.
계정 크레딧이 부족합니다. 계정 잔액을 확인하세요. 추가 작업을 제출하기 전에 크레딧을 구매하거나 다음 플랜 크레딧 지급 시점까지 기다리세요.
입력 필드와 선택한 모드가 일치하지 않습니다. name_company에서는 이름 필드를 분리하거나 fullName을 사용하고, linkedin에서는 linkedInProfile을 사용하세요. 연락처를 수정한 뒤 하나의 일관된 모드로 제출하세요.
작업 상태가 failed 또는 stopped입니다. 스크립트는 작업 ID, 최종 상태, 확인 가능한 작업 오류가 포함된 RuntimeError를 발생시킵니다. 어떤 연락처를 다시 조회할지 결정하기 전에 해당 작업과 반환된 결과를 확인하세요.
POST 응답이 유실되었습니다. 제출 전에 출력된 uniqueId를 복사하세요. 같은 키로 작업 목록을 조회합니다.
curl "https://api.datablist.com/v1/enrichments/tasks" \
-H "X-API-Key: $DATABLIST_API_KEY"출력된 uniqueId와 일치하는 작업을 찾아 @id를 복사한 뒤 작업 상태 또는 결과 URL을 사용해 계속 진행하세요. 연락처를 자동으로 다시 제출하지 않고 기존 작업을 복구할 수 있습니다. 작업 수명 주기와 결과 Endpoint는 작업 API 문서에서 확인할 수 있습니다.
다음 단계
이제 지원되는 세 가지 입력 방식에 사용할 수 있는 하나의 blocking Python Requests 함수와, 찾아서 인증한 이메일을 캠페인 검토용으로 준비하는 로컬 필터를 갖추었습니다.
예제 연락처 두 건을 직접 조사한 연락처 중 일부로 교체해 보세요. 데이터에 맞는 입력 방식을 선택해 조회를 실행하고, 이메일을 사용하기 전에 emailStatus를 검토하세요. 더 큰 목록을 처리하는 방법은 대규모 이메일 찾기 가이드를 참고하세요. 관리되는 GitHub 저장소에서 다운로드 가능한 blocking 및 비동기 Python 예제를 확인할 수 있습니다.