当前位置：首页 > 学习笔记 > 正文内容

Notion API 自动化工作流实战教程：从零打造高效数字工作台

廖万里10小时前学习笔记1

Notion API 自动化工作流

# Notion API 自动化工作流实战教程：从零打造高效数字工作台

本文将系统讲解如何利用 Notion API 构建自动化工作流，从 API 认证基础到 Python 脚本操作，再到 n8n 可视化工作流搭建，带你一步步实现数据库自动更新、表单数据同步等实战场景，打造真正高效的数字化工作系统。

## 一、Notion API 基础认知 Notion 在 2021 年开放了公开 API，让这款笔记工具真正具备了成为"应用构建平台"的潜力。通过 API，你可以实现数据的自动读取、写入、更新和删除，将 Notion 从被动的信息存储工具转变为主动的工作流引擎。 ### 1.1 API 的核心概念 Notion API 基于 REST 架构，采用 JSON 格式传输数据。理解以下概念是入门的第一步： - **Integration（集成）**：连接外部应用与 Notion 工作区的桥梁 - **Database（数据库）**：结构化数据存储，支持属性和视图 - **Page（页面）**：内容容器，可包含文本、数据库等 - **Block（块）**：内容的最小单元，如段落、标题、代码块等 ### 1.2 权限模型 Notion 采用"显式授权"模式，集成默认没有访问权限，需要用户手动将页面或数据库共享给集成。这种设计保证了安全性，但也意味着你需要在每次使用新数据库时进行授权操作。 ## 二、集成设置与认证配置 ### 2.1 创建 Notion Integration 第一步是在 Notion 开发者平台创建集成： 1. 访问 https://www.notion.so/my-integrations 2. 点击"New integration" 3. 填写名称、选择工作区 4. 获取 Internal Integration Token（密钥）

# .env 文件存储敏感信息
NOTION_API_KEY = "secret_..."
DATABASE_ID = "..."

### 2.2 授权数据库访问在 Notion 中打开目标数据库页面，点击右上角"..."→"Add connections"→选择你的集成。这一步至关重要，未经授权的 API 调用将返回 403 错误。 ### 2.3 Python 环境准备推荐使用官方 SDK `notion-client`，简化 HTTP 请求的封装：

# 安装依赖
pip install notion-client python-dotenv

# 初始化客户端
from notion_client import Client
import os
from dotenv import load_dotenv

load_dotenv()

notion = Client(auth=os.getenv("NOTION_API_KEY"))

## 三、Python 操作 Notion 实战 ### 3.1 查询数据库内容

def query_database(database_id, filter_params=None):
    """查询 Notion 数据库"""
    response = notion.databases.query(
        database_id=database_id,
        filter=filter_params
    )
    
    results = response["results"]
    
    # 处理分页
    while response["has_more"]:
        response = notion.databases.query(
            database_id=database_id,
            filter=filter_params,
            start_cursor=response["next_cursor"]
        )
        results.extend(response["results"])
    
    return results

# 示例：查询状态为"进行中"的任务
tasks = query_database(
    database_id=os.getenv("DATABASE_ID"),
    filter_params={
        "property": "Status",
        "select": {"equals": "进行中"}
    }
)

### 3.2 创建新页面（添加记录）

def create_page(database_id, title, properties=None):
    """在数据库中创建新页面"""
    page_data = {
        "parent": {"database_id": database_id},
        "properties": {
            "Name": {
                "title": [{"text": {"content": title}}]
            }
        }
    }
    
    # 添加额外属性
    if properties:
        page_data["properties"].update(properties)
    
    return notion.pages.create(**page_data)

# 示例：创建任务记录
create_page(
    database_id=os.getenv("DATABASE_ID"),
    title="完成 Notion API 教程",
    properties={
        "Status": {"select": {"name": "待处理"}},
        "Priority": {"select": {"name": "高"}},
        "Tags": {"multi_select": [{"name": "学习"}]}
    }
)

### 3.3 更新页面属性

def update_page_status(page_id, new_status):
    """更新页面状态属性"""
    notion.pages.update(
        page_id=page_id,
        properties={
            "Status": {"select": {"name": new_status}}
        }
    )

# 示例：将任务标记为完成
update_page_status(page_id="...", new_status="已完成")

### 3.4 添加内容块

def append_content(page_id, text):
    """向页面追加文本块"""
    notion.blocks.children.append(
        block_id=page_id,
        children=[
            {
                "type": "paragraph",
                "paragraph": {
                    "rich_text": [{"text": {"content": text}}]
                }
            }
        ]
    )

## 四、数据库自动化实战场景 ### 4.1 场景一：定时同步外部数据到 Notion 假设你需要将 GitHub Issues 同步到 Notion 进行统一管理：

import requests
from datetime import datetime

def sync_github_issues(repo_owner, repo_name, database_id):
    """同步 GitHub Issues 到 Notion"""
    url = f"https://api.github.com/repos/{repo_owner}/{repo_name}/issues"
    response = requests.get(url)
    issues = response.json()
    
    for issue in issues:
        # 检查是否已存在
        existing = query_database(
            database_id=database_id,
            filter_params={
                "property": "Issue URL",
                "url": {"equals": issue["html_url"]}
            }
        )
        
        if not existing:
            create_page(
                database_id=database_id,
                title=issue["title"],
                properties={
                    "Issue URL": {"url": issue["html_url"]},
                    "Status": {"select": {"name": "Open" if issue["state"] == "open" else "Closed"}},
                    "Created": {"date": {"start": issue["created_at"][:10]}}
                }
            )

### 4.2 场景二：自动化周报生成

def generate_weekly_report(database_id):
    """生成本周工作总结"""
    from datetime import datetime, timedelta
    
    week_start = (datetime.now() - timedelta(days=datetime.now().weekday())).strftime("%Y-%m-%d")
    
    completed_tasks = query_database(
        database_id=database_id,
        filter_params={
            "and": [
                {"property": "Status", "select": {"equals": "已完成"}},
                {"property": "Completed Date", "date": {"on_or_after": week_start}}
            ]
        }
    )
    
    report_lines = [f"## 本周完成 ({len(completed_tasks)} 项)\n"]
    
    for task in completed_tasks:
        title = task["properties"]["Name"]["title"][0]["text"]["content"]
        report_lines.append(f"- {title}")
    
    return "\n".join(report_lines)

### 4.3 场景三：跨数据库关联更新

def sync_related_databases(source_db_id, target_db_id):
    """跨数据库关联更新"""
    # 从源数据库获取所有项目
    projects = query_database(database_id=source_db_id)
    
    for project in projects:
        project_name = project["properties"]["Name"]["title"][0]["text"]["content"]
        project_id = project["id"]
        
        # 查询关联任务
        tasks = query_database(
            database_id=target_db_id,
            filter_params={
                "property": "Project",
                "relation": {"contains": project_id}
            }
        )
        
        # 更新项目的任务计数
        notion.pages.update(
            page_id=project_id,
            properties={
                "Task Count": {"number": len(tasks)}
            }
        )

## 五、表单集成：让外部数据自动流入 Notion Notion 原生不支持表单功能，但通过 API 可以轻松实现。以下是结合 Typeform 的自动化方案： ### 5.1 Webhook 接收器

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/webhook/typeform", methods=["POST"])
def handle_typeform():
    """处理 Typeform 表单提交"""
    data = request.json
    
    # 提取表单字段
    form_data = {}
    for answer in data["form_response"]["answers"]:
        field_id = answer["field"]["id"]
        if answer["type"] == "text":
            form_data[field_id] = answer["text"]
        elif answer["type"] == "choice":
            form_data[field_id] = answer["choice"]["label"]
    
    # 写入 Notion
    create_page(
        database_id=os.getenv("FORM_DATABASE_ID"),
        title=form_data.get("name", "New Submission"),
        properties={
            "Email": {"email": form_data.get("email")},
            "Source": {"select": {"name": "Typeform"}},
            "Submitted At": {"date": {"start": datetime.now().isoformat()}}
        }
    )
    
    return jsonify({"status": "success"}), 200

### 5.2 Google Forms 集成方案通过 Google Apps Script 作为中间层：

# Google Apps Script 代码（需部署为 Web App）
function doPost(e) {
  var formData = JSON.parse(e.postData.contents);
  
  var notionPayload = {
    parent: {database_id: "YOUR_DATABASE_ID"},
    properties: {
      Name: {title: [{text: {content: formData.name}}]},
      Email: {email: formData.email}
    }
  };
  
  var options = {
    method: "post",
    contentType: "application/json",
    headers: {
      "Authorization": "Bearer YOUR_NOTION_TOKEN",
      "Notion-Version": "2022-06-28"
    },
    payload: JSON.stringify(notionPayload)
  };
  
  UrlFetchApp.fetch("https://api.notion.com/v1/pages", options);
}

## 六、n8n 可视化工作流搭建对于不熟悉编程的用户，n8n 提供了拖拽式的工作流搭建界面，让自动化触手可及。 ### 6.1 n8n 核心优势 - **可视化编辑**：节点拖拽连接，逻辑一目了然 - **丰富的集成**：内置 400+ 应用节点，Notion 原生支持 - **自托管免费**：可部署在本地或私有服务器 - **代码扩展**：支持 JavaScript 自定义节点 ### 6.2 实战：构建"表单提交 → Notion 记录 → 邮件通知"工作流 #### 步骤一：添加 Webhook 触发器在 n8n 中创建新工作流，添加 Webhook 节点作为触发器。选择 POST 方法，复制生成的 Webhook URL。 #### 步骤二：配置 Notion 节点添加 Notion 节点，选择"Create Page"操作： - **认证**：使用 API Key 或 OAuth - **数据库**：选择目标数据库 - **属性映射**：将 Webhook 数据映射到 Notion 属性 #### 步骤三：添加邮件通知添加 Email 节点，配置 SMTP 信息，使用模板语法引用 Notion 创建的数据： ``` 主题：新表单提交 - {{ $json.properties.Name.title[0].text.content }} 正文：收到新提交：姓名：{{ $json.properties.Name.title[0].text.content }} 邮箱：{{ $json.properties.Email.email }} ``` ### 6.3 进阶：定时同步工作流构建每天自动同步数据的工作流： 1. **Schedule Trigger**：设置 Cron 表达式（如 `0 9 * * *` 每天 9 点执行） 2. **HTTP Request**：调用外部 API 获取数据 3. **Function Node**：数据转换处理 4. **Notion Node**：批量写入数据库

# n8n Function Node 示例代码
const items = [];

for (const record of $input.all()) {
  items.push({
    json: {
      title: record.json.name,
      properties: {
        Status: {select: {name: "Active"}},
        Score: {number: record.json.score}
      }
    }
  });
}

return items;

### 6.4 错误处理与重试在工作流中添加 Error Trigger 节点，捕获失败任务并记录到专门的 Notion 数据库：

// 错误处理逻辑
{
  "workflow": "{{ $workflow.name }}",
  "error": "{{ $json.error.message }}",
  "timestamp": "{{ $now }}",
  "execution_id": "{{ $execution.id }}"
}

## 七、最佳实践与注意事项 ### 7.1 性能优化 - **批量操作**：使用 `bulk` 端点减少 API 调用次数 - **分页处理**：大数据集务必处理分页，避免内存溢出 - **缓存策略**：对频繁访问的配置数据进行本地缓存 - **速率限制**：Notion API 有请求频率限制（3 requests/second），需要合理控制 ### 7.2 安全建议 - **密钥管理**：绝不将 API Key 硬编码在代码中 - **最小权限原则**：仅授权必要的数据库访问权限 - **日志脱敏**：记录日志时过滤敏感信息 - **定期轮换**：定期更新 Integration Token ### 7.3 常见问题排查 | 错误代码 | 原因 | 解决方案 | |---------|------|---------| | 401 | 认证失败 | 检查 API Key 是否正确 | | 403 | 权限不足 | 确认数据库已授权给集成 | | 404 | 资源不存在 | 检查 Database ID 格式 | | 429 | 请求过多 | 实现退避重试机制 | ### 7.4 调试技巧

import logging

# 配置详细日志
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

def debug_api_call(func):
    """API 调用装饰器"""
    def wrapper(*args, **kwargs):
        try:
            result = func(*args, **kwargs)
            logger.debug(f"API Success: {func.__name__}")
            return result
        except Exception as e:
            logger.error(f"API Error: {func.__name__} - {str(e)}")
            raise
    return wrapper

## 八、总结 Notion API 为知识工作者打开了一扇通往自动化的大门。从基础的 Python 脚本到可视化的 n8n 工作流，你可以根据技术能力和场景需求选择合适的实现方式。关键要点回顾： 1. **理解认证模型**：Integration + 显式授权是安全的基础 2. **善用官方 SDK**：notion-client 大幅降低开发难度 3. **从简单场景切入**：建议先实现单一功能的自动化，再逐步扩展 4. **重视错误处理**：生产环境必须有完善的异常捕获机制 5. **平衡定制化与可维护性**：代码越简单，后期维护成本越低 Notion 的 API 生态正在快速发展，配合 n8n、Make 等工具，你可以构建出媲美专业系统的自动化工作流。重要的是开始动手实践——从一个小需求出发，逐步掌握 API 的使用技巧，最终打造属于自己的高效数字工作台。记住：自动化的目标是释放重复劳动的时间，让你专注于真正有价值的工作。愿你在 Notion 自动化之路上越走越远，效率翻倍。

本文链接：https://www.kkkliao.cn/?id=953 转载需授权！

分享到：