时间戳增量同步需用timezone-aware的datetime.now(timezone.utc)、状态存数据库sync_state表、查询带上限+排序+去重兜底,确保幂等可恢复。

时间戳增量同步不是“查大于上次时间”就完事——漏数据、重复写、时钟漂移这三类问题,90% 都出在时区混用和状态存储不可靠上。
datetime.utcnow() 和 datetime.now(timezone.utc) 的区别必须分清
Python 里 datetime.utcnow() 返回的是 timezone-naive 对象,它不带时区信息,但值是 UTC 时间;而 datetime.now(timezone.utc) 返回的是 timezone-aware 对象,明确标记为 UTC。数据库(如 PostgreSQL)在做 WHERE updated_at >= %s 比较时,如果传入的是 naive 时间,DBMS 可能按本地时区解释或静默转成 UTC,结果不可控。
- 永远别用
datetime.now()或datetime.utcnow()直接生成同步起点 - 统一用
datetime.now(timezone.utc)(需from datetime import timezone) - 查库前实时取当前时间:不是脚本启动时取一次,而是每次 SQL 查询前调用一次
- SQLAlchemy 中读取字段时,显式声明类型:
type_=postgresql.TIMESTAMP(timezone=True)
last_sync_time 不能存在本地文件或内存里
容器重启、多实例并行、NFS 缓存延迟,都会让文件里的时间戳失效。哪怕只跑一个进程,也得考虑异常中断后状态丢失的问题。
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac,无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启:辅助功能 + 屏幕录制权限
- 必须建一张
sync_state表,字段至少含table_name、last_sync_time、updated_at(自动更新) - 写入要用
INSERT ... ON CONFLICT (table_name) DO UPDATE(PostgreSQL)或REPLACE INTO(MySQL),避免竞态 - 读取时加
SELECT ... FOR UPDATE锁住该行,防止两个同步任务同时读到同一时间戳 - 不要依赖配置文件或环境变量硬编码初始时间,首次同步前先执行
SELECT MAX(updated_at) FROM my_table
增量查询 SQL 必须带“上限时间 + 排序 + 去重兜底”
只写 WHERE updated_at > %(last_time)s 是危险的:并发写入可能让多条记录拥有相同 updated_at;系统时钟回拨会让新写入的记录时间戳小于上次同步点;批量更新语句也可能导致一批记录时间戳完全一致。
- 查询条件应为:
WHERE updated_at >= %(last_time)s AND updated_at - 必须
ORDER BY updated_at, id,确保分页稳定(仅靠updated_at不够) - 更稳妥的做法是用复合比较:
WHERE (updated_at, id) > (%(last_time)s, %(last_id)s) - 每次查询后,用
SELECT MAX(updated_at), MAX(id) FROM ...更新状态,而不是用 Python max()
真正难的不是写对一条 SQL,而是让整个同步链路在时钟漂移、网络中断、进程崩溃、多实例抢占这些现实条件下仍保持幂等和可恢复——所有状态变更都要原子化,所有时间戳都要带时区标记,所有“上次”都要从数据库里实时读,而不是从变量或文件里猜。
Python免费学习笔记(深入):立即使用
在学习笔记中,你将探索 Python 的核心概念和高级技巧!










