第四章：数据源连接——打通迁移的“第一公里”

写迁移脚本，最核心的任务其实是跟数据打交道。数据在哪？怎么把它读进来？这就是本章要解决的问题。

我个人习惯把数据源分成三大类：数据库、文件、API。听上去简单吧？但每个大类里的坑，还真不少。我曾经在一个项目里，就因为搞错了CSV文件的编码格式，导致整个线上迁移跑出了一堆乱码，被运营同事追着问了好几天。

下面我们就来逐一拆解。

数据库连接：稳定是第一位

数据库是迁移脚本最常打交道的对象。不管是 MySQL、PostgreSQL 还是 Oracle，连接方式的核心逻辑都一样——用驱动建立连接，然后执行 SQL。

但这里有个误区：很多人觉得连上数据库就完事了。我建议你把注意力放在「连接管理」上。

对比维度	短连接（每次新建）	连接池（复用）
性能	差，建立连接很慢（TCP握手套接字）	好，复用已有连接
资源占用	高，频繁创建销毁	可控，固定数量
适用场景	一次性小数据量迁移	大规模持续迁移
实现复杂度	简单	稍复杂，需引入池化库

我的习惯：只要脚本预期运行超过 5 分钟，我一定用连接池。Python 的 SQLAlchemy 自带池化能力，Java 有 HikariCP，别自己手写连接池——坑太多了。

举个例子，用 Python 连接 MySQL 的基本代码长这样：

import pymysql
from dbutils.pooled_db import PooledDB

# 创建连接池
pool = PooledDB(
    creator=pymysql,
    maxconnections=10,
    host='10.0.0.1',
    port=3306,
    user='migrator',
    password='secret',
    database='source_db'
)

# 从池中获取连接
conn = pool.connection()
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE updated_at >= %s", (last_run,))
rows = cursor.fetchall()
cursor.close()
conn.close()  # 不是真关闭，是归还到池里

这里我特别强调一下 参数化查询。直接拼接 SQL 字符串？千万别。我见到过因为一个单引号没处理好，整个迁移脚本原地爆炸的例子。用 %s 或 ? 占位符，既安全又省心。

文件连接：看似简单，实则细节多

很多人觉得读文件是基本功，不值得专门说。但你真的处理好所有情况了吗？

CSV 文件的编码、分隔符、引号、换行符——每个都是潜在的地雷。JSON 文件的嵌套深度、数组结构、大文件拆分——处理不好就 OOM。

我一般这样处理文件连接：

路径管理：用配置文件管理路径，路径或相对于项目根目录。别硬编码。
编码检测：用 chardet 库检测编码，或者统一要求 UTF-8 + BOM。
流式读取：对于大文件，一定用迭代器逐行处理，别一次性全读进内存。
格式解析：CSV 用 csv.reader，JSON 用 ijson 流式解析，XML 用 xml.etree。

来看看我最常用的 CSV 读取模式：

import csv
import chardet

# 先检测编码
with open('data.csv', 'rb') as f:
    raw = f.read(10000)
    encoding = chardet.detect(raw)['encoding']

# 再用检测到的编码打开
with open('data.csv', 'r', encoding=encoding) as f:
    reader = csv.DictReader(f)
    for row in reader:
        # 处理每一行数据
        process_row(row)

我曾经踩过的坑：Excel 导出的 CSV 文件很可能带 BOM 头。Python 的 csv 模块读到 BOM 时，第一列列名会变成 "\ufeffcolumn_name"。我的解决方法是统一用 utf-8-sig 编码读取，它会自动处理 BOM。

API 连接：稳定性和限流是关键

现在越来越多的系统暴露 RESTful 或 gRPC 接口。迁移脚本通过 API 拉取数据，听起来很优雅——但实际执行时，你会遇到各种问题。

最常见的三大问题是：认证、限流、重试。

认证

API 通常需要或。我习惯把认证信息放在环境变量或加密的配置文件中。千万别把 key 写死在代码里——之前有个同事把密钥提交到了 Git 仓库，后果很严重。

import os
import requests

 = os.environ.get('SOURCE_')
BASE_URL = 'https://api.example.com/v1'

headers = {
    'Authorization': f'Bearer {}',
    'Content-Type': 'application/json'
}

response = requests.get(f'{BASE_URL}/users', headers=headers)
data = response.json()

限流处理

很多 API 有调用频率限制。比如每秒最多 10 次请求。你不处理这个，跑着跑着就被 429 状态码打回来了。

我的做法是用 指数退避 + 随机抖动：

import time
import random

def api_call_with_retry(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 429:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"触发限流，等待 {wait:.2f} 秒后重试")
            time.sleep(wait)
        else:
            response.raise_for_status()

这里有个细节：指数退避的指数底数我一般选 2，初始值 1 秒。最多重试 5 次。如果 5 次都失败，就放弃这条记录并写入错误日志——别让单个 API 的失败拖垮整个迁移流水线。

重试与幂等

API 调用可能因为网络抖动失败，也可能成功但响应超时。所以你的重试逻辑一定要设计成幂等的。也就是说，同一个请求执行多次，效果跟执行一次一样。

说实话，很多系统设计者没考虑到这点。我建议你在调用 API 前先记录一个唯一请求 ID，服务端如果收到重复 ID 就直接返回上次的结果。

连接配置的外部化

无论你用的是数据库、文件还是 API，连接信息都不应该写在脚本里。这个道理大家都懂，但实际项目里我见过太多硬编码的例子。

我的标准做法是：

用 YAML 或 TOML 文件管理连接配置
敏感信息（密码、）用环境变量注入
不同环境（开发、测试、生产）使用不同的配置文件

一份典型的配置文件长这样：

# config.yaml
source:
  type: mysql
  host: ${MYSQL_HOST}
  port: 3306
  user: ${MYSQL_USER}
  password: ${MYSQL_PASS}
  database: source_db

output:
  type: json_file
  path: /data/output/
  encoding: utf-8

api:
  url: https://api.example.com/v2
  key: ${}
  rate_limit: 100

我个人习惯：用 python-dotenv 库加载 .env 文件到环境变量，然后在配置文件中通过 ${VARIABLE} 语法引用。这样既保持配置文件的整洁，又把敏感信息隔离了出去。

好，关于数据源连接就聊这么多。这三种方式各有各的脾气：数据库要管好连接池，文件要小心编码和内存，API 要处理好限流和重试。掌握这些，你的迁移脚本才算真正打通了任督二脉。