파이썬으로 노션 데이터베이스에 CSV 파일 업로드하는 방법

노션 API와 파이썬을 활용하면 CSV 파일의 데이터를 노션 데이터베이스에 자동으로 업로드할 수 있어요. Zapier나 다른 유료 서비스 없이도 무료로 구현 가능한 방법을 알아볼게요.

노션 API 토큰 발급받기

먼저 노션 개발자 페이지에서 API 토큰을 발급받아야 해요. 브라우저에서 notion.so/my-integrations에 접속한 후 새 통합을 만들어주세요.

통합 생성 시 이름을 입력하고 작업 공간을 선택하면 API 토큰이 생성돼요. 이 토큰은 나중에 파이썬 코드에서 사용하니 안전한 곳에 복사해 두세요.

노션 데이터베이스 준비하기

CSV 데이터를 받을 노션 데이터베이스를 만들어야 해요. 데이터베이스 생성 후 우측 상단의 점 세 개 메뉴를 클릭하고 연결 항목에서 방금 만든 통합을 추가해주세요.

데이터베이스 URL에서 ID를 확인할 수 있어요. 브라우저 주소창에서 notion.so/ 다음에 나오는 32자리 문자열이 데이터베이스 ID예요.

https://www.notion.so/myworkspace/a1b2c3d4e5f6... 
# a1b2c3d4e5f6... 부분이 데이터베이스 ID

필요한 패키지 설치하기

터미널을 열고 필요한 파이썬 패키지를 설치해요.

pip install pandas requests

pandas는 CSV 파일을 읽기 위해, requests는 노션 API와 통신하기 위해 사용돼요.

기본 업로드 코드 작성하기

이제 파이썬 코드를 작성해볼게요. 새 파일을 만들고 다음 코드를 입력하세요.

import pandas as pd
import requests
import json
import time

# 노션 API 설정
NOTION_API_KEY = 'secret_여기에_API_토큰_입력'
DATABASE_ID = '여기에_데이터베이스_ID_입력'

headers = {
    "Authorization": f"Bearer {NOTION_API_KEY}",
    "Content-Type": "application/json",
    "Notion-Version": "2022-06-28"
}

# CSV 파일 읽기
df = pd.read_csv('data.csv', encoding='utf-8')

# 각 행을 노션에 업로드
for index, row in df.iterrows():
    # 노션 페이지 생성을 위한 데이터 구조
    data = {
        "parent": {"database_id": DATABASE_ID},
        "properties": {
            "이름": {
                "title": [{"text": {"content": str(row['이름'])}}]
            },
            "이메일": {
                "email": str(row['이메일'])
            },
            "전화번호": {
                "phone_number": str(row['전화번호'])
            }
        }
    }
    
    # API 요청 전송
    response = requests.post(
        'https://api.notion.com/v1/pages',
        headers=headers,
        data=json.dumps(data)
    )
    
    if response.status_code == 200:
        print(f"{index + 1}번째 데이터 업로드 성공")
    else:
        print(f"{index + 1}번째 데이터 업로드 실패: {response.text}")
    
    # API 제한 방지를 위한 대기
    time.sleep(0.3)

다양한 속성 타입 처리하기

노션 데이터베이스는 여러 속성 타입을 지원해요. 각 타입별로 데이터 구조가 다르니 주의해야 해요.

# 다양한 속성 타입 예제
properties = {
    # 제목 (Title)
    "제목": {
        "title": [{"text": {"content": str(row['제목'])}}]
    },
    
    # 텍스트 (Rich Text)
    "설명": {
        "rich_text": [{"text": {"content": str(row['설명'])}}]
    },
    
    # 숫자 (Number)
    "가격": {
        "number": float(row['가격'])
    },
    
    # 선택 (Select)
    "카테고리": {
        "select": {"name": str(row['카테고리'])}
    },
    
    # 다중 선택 (Multi-select)
    "태그": {
        "multi_select": [
            {"name": tag.strip()} 
            for tag in str(row['태그']).split(',')
        ]
    },
    
    # 날짜 (Date)
    "생성일": {
        "date": {"start": str(row['생성일'])}
    },
    
    # 체크박스 (Checkbox)
    "완료": {
        "checkbox": row['완료'] == 'True'
    },
    
    # URL
    "링크": {
        "url": str(row['링크'])
    }
}

중복 데이터 방지하기

노션 API는 중복 체크를 자동으로 하지 않아요. 중복을 방지하려면 업로드 전에 기존 데이터를 조회해야 해요.

def check_duplicate(email):
    """이메일로 중복 확인"""
    query_data = {
        "filter": {
            "property": "이메일",
            "email": {
                "equals": email
            }
        }
    }
    
    response = requests.post(
        f'https://api.notion.com/v1/databases/{DATABASE_ID}/query',
        headers=headers,
        data=json.dumps(query_data)
    )
    
    if response.status_code == 200:
        results = response.json()
        return len(results['results']) > 0
    return False

# 중복 체크를 포함한 업로드
for index, row in df.iterrows():
    # 중복 확인
    if check_duplicate(row['이메일']):
        print(f"{row['이름']} - 이미 존재하는 데이터입니다")
        continue
    
    # 데이터 업로드 코드...

에러 처리와 재시도 로직

네트워크 문제나 API 제한으로 업로드가 실패할 수 있어요. 재시도 로직을 추가하면 안정성이 높아져요.

def upload_with_retry(data, max_retries=3):
    """재시도 로직이 포함된 업로드 함수"""
    for attempt in range(max_retries):
        try:
            response = requests.post(
                'https://api.notion.com/v1/pages',
                headers=headers,
                data=json.dumps(data),
                timeout=10
            )
            
            if response.status_code == 200:
                return True, "성공"
            elif response.status_code == 429:
                # Rate limit 초과
                wait_time = int(response.headers.get('Retry-After', 60))
                print(f"Rate limit 초과. {wait_time}초 대기...")
                time.sleep(wait_time)
            else:
                error_msg = response.json().get('message', '알 수 없는 오류')
                return False, error_msg
                
        except requests.exceptions.RequestException as e:
            print(f"네트워크 오류 (시도 {attempt + 1}/{max_retries}): {e}")
            if attempt < max_retries - 1:
                time.sleep(5)
    
    return False, "최대 재시도 횟수 초과"

대용량 CSV 처리하기

CSV 파일이 크면 메모리 문제가 발생할 수 있어요. 청크 단위로 읽어서 처리하는 방법을 사용하세요.

# 대용량 CSV를 청크로 처리
chunk_size = 100
total_uploaded = 0
failed_rows = []

for chunk in pd.read_csv('large_data.csv', chunksize=chunk_size):
    for index, row in chunk.iterrows():
        # 데이터 준비
        data = {
            "parent": {"database_id": DATABASE_ID},
            "properties": {
                # 속성 매핑...
            }
        }
        
        # 업로드 시도
        success, message = upload_with_retry(data)
        
        if success:
            total_uploaded += 1
            print(f"진행률: {total_uploaded}개 업로드 완료")
        else:
            failed_rows.append({
                'row_index': index,
                'data': row.to_dict(),
                'error': message
            })

# 실패한 데이터 저장
if failed_rows:
    pd.DataFrame(failed_rows).to_csv('failed_uploads.csv', index=False)
    print(f"{len(failed_rows)}개 업로드 실패. failed_uploads.csv 확인")

진행 상황 표시하기

업로드 진행 상황을 시각적으로 보여주면 사용성이 좋아져요.

from tqdm import tqdm

# tqdm 설치: pip install tqdm

# 진행률 표시와 함께 업로드
total_rows = len(df)
with tqdm(total=total_rows, desc="업로드 진행") as pbar:
    for index, row in df.iterrows():
        # 데이터 업로드 코드...
        
        pbar.update(1)
        pbar.set_postfix({
            '성공': total_uploaded,
            '실패': len(failed_rows)
        })

전체 자동화 스크립트

모든 기능을 통합한 완성된 스크립트예요.

import pandas as pd
import requests
import json
import time
from datetime import datetime
import os

class NotionCSVUploader:
    def __init__(self, api_key, database_id):
        self.api_key = api_key
        self.database_id = database_id
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "Notion-Version": "2022-06-28"
        }
        self.base_url = "https://api.notion.com/v1"
        
    def upload_csv(self, csv_file, property_mapping):
        """CSV 파일을 노션 데이터베이스에 업로드"""
        # CSV 읽기
        df = pd.read_csv(csv_file, encoding='utf-8')
        print(f"총 {len(df)}개 데이터 발견")
        
        success_count = 0
        failed_rows = []
        
        for index, row in df.iterrows():
            # 노션 속성 구성
            properties = {}
            for csv_column, notion_config in property_mapping.items():
                if csv_column in row:
                    prop_type = notion_config['type']
                    prop_name = notion_config['name']
                    value = row[csv_column]
                    
                    # 타입별 처리
                    if prop_type == 'title':
                        properties[prop_name] = {
                            "title": [{"text": {"content": str(value)}}]
                        }
                    elif prop_type == 'rich_text':
                        properties[prop_name] = {
                            "rich_text": [{"text": {"content": str(value)}}]
                        }
                    elif prop_type == 'number':
                        properties[prop_name] = {
                            "number": float(value) if pd.notna(value) else None
                        }
                    elif prop_type == 'select':
                        properties[prop_name] = {
                            "select": {"name": str(value)}
                        }
                    elif prop_type == 'email':
                        properties[prop_name] = {
                            "email": str(value)
                        }
            
            # 페이지 생성
            data = {
                "parent": {"database_id": self.database_id},
                "properties": properties
            }
            
            try:
                response = requests.post(
                    f"{self.base_url}/pages",
                    headers=self.headers,
                    data=json.dumps(data)
                )
                
                if response.status_code == 200:
                    success_count += 1
                    print(f"✓ {index + 1}/{len(df)} 업로드 성공")
                else:
                    failed_rows.append({
                        'index': index,
                        'error': response.text
                    })
                    print(f"✗ {index + 1}/{len(df)} 업로드 실패")
                
                # API 제한 방지
                time.sleep(0.3)
                
            except Exception as e:
                failed_rows.append({
                    'index': index,
                    'error': str(e)
                })
        
        # 결과 저장
        print(f"\n업로드 완료: 성공 {success_count}개, 실패 {len(failed_rows)}개")
        
        if failed_rows:
            error_file = f"upload_errors_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
            with open(error_file, 'w', encoding='utf-8') as f:
                json.dump(failed_rows, f, ensure_ascii=False, indent=2)
            print(f"오류 내역이 {error_file}에 저장되었습니다")

# 사용 예제
if __name__ == "__main__":
    # 설정
    API_KEY = 'your_api_key_here'
    DATABASE_ID = 'your_database_id_here'
    
    # 속성 매핑 정의
    mapping = {
        'name': {'name': '이름', 'type': 'title'},
        'email': {'name': '이메일', 'type': 'email'},
        'phone': {'name': '전화번호', 'type': 'rich_text'},
        'amount': {'name': '금액', 'type': 'number'},
        'status': {'name': '상태', 'type': 'select'}
    }
    
    # 업로더 생성 및 실행
    uploader = NotionCSVUploader(API_KEY, DATABASE_ID)
    uploader.upload_csv('data.csv', mapping)

이렇게 파이썬과 노션 API를 활용하면 CSV 데이터를 효율적으로 업로드할 수 있어요. 코드를 수정해서 여러분의 데이터베이스 구조에 맞게 활용해보세요.

힙랩

이 블로그 검색