Python 读取数据库 BLOB 字段并落地文件：从连接配置到本地存储的完整实践

1. 从一次真实需求说起Python 读取数据库 BLOB 字段并落地文件到底难在哪先说清楚这件事是什么。数据库里的 BLOB 字段本质就是一段二进制大对象图片、PDF、Word、音频、压缩包都能塞进去。Python 读取数据库 BLOB 字段并落地文件就是把这段二进制从结果集里取出来原样写到本地磁盘得到一个能双击打开的文件。适合谁做数据迁移的后端、要把历史附件导出的运维、做数据清洗的数据工程师以及需要批量把库里图片还原成文件目录的开发者。听起来简单但真正动手时坑不少。我见过太多人卡在几个地方一是驱动选错Oracle 用 cx_Oracle、MySQL 用 pymysql 或 mysqlclient、PostgreSQL 用 psycopg2、SQL Server 用 pyodbc连接串写法完全不同二是 BLOB 读出来是 LOB 对象而不是 bytes直接 write 会报类型错误三是字符集没设对中文路径或中文内容乱码四是文件写了一半程序崩了得到一个损坏的半截文件五是大字段一次性读进内存几百 MB 直接把进程撑爆。还有一个容易被忽略的点很多团队现在不直接连生产库而是走一层统一 API 通道来访问数据服务。这样做的好处是连接参数集中管理、权限可控、审计清晰。所以这篇不只讲直连数据库也会讲怎么通过统一 API 通道把 BLOB 取回来再落地两条路都给你可复制的代码。核心检索词先摆出来Python 读取数据库 BLOB 字段并存储为本地文件。下面从连接配置一路走到完整性校验每一步都能直接抄。我会用 Oracle 作为主例子因为原场景就是它同时给出 MySQL、PostgreSQL 的对照写法你按自己的库替换即可。先明确整体流程心里有个地图建立连接 → 执行查询 → 逐行取出 BLOB → 转成 bytes → 写入本地文件 → 校验大小或哈希 → 关闭游标和连接。六步缺一不可。很多人只做前三步就以为完事了结果文件打不开回头查半天。我们一步步来。2. 前置准备驱动安装、连接参数与统一 API 通道的接入方式动手之前先把环境铺好。这一步做扎实后面能省掉一大半排错时间。2.1 驱动安装与版本对应不同数据库的 Python 驱动不一样装错了连接都建不起来。对照表如下数据库推荐驱动安装命令备注Oraclecx_Oracle / oracledbpip install oracledb新版官方推荐 oracledb免装 Instant Client 的 thin 模式MySQLPyMySQLpip install pymysql纯 Python无需编译PostgreSQLpsycopg2-binarypip install psycopg2-binarybinary 版免编译SQL Serverpyodbcpip install pyodbc需系统装 ODBC DriverOracle 这里特别说一下老代码用 cx_Oracle需要本地装 Instant Client 并配环境变量新项目建议直接用 oracledb 的 thin 模式纯 Python 实现不用折腾客户端库。如果你维护的是老系统cx_Oracle 也能跑只是部署时记得把客户端带上。2.2 连接参数怎么组织连接参数别硬编码在脚本里用字典或环境变量管理。一个典型的 Oracle 连接串是「用户名/密码主机:端口/服务名」。我习惯这样组织import os DB_CONFIG { user: os.getenv(DB_USER, your_user), password: os.getenv(DB_PASSWORD, your_password), dsn: os.getenv(DB_DSN, 10.0.0.10:1521/orcl), }MySQL 则是 host、port、user、password、database、charset 分开写。PostgreSQL 多一个 schema。把这些抽出来换环境只改配置不改代码。2.3 走统一 API 通道的接入方式如果你的团队不允许直连数据库而是通过统一 API 通道访问数据服务那接入方式会变成 HTTP 请求。这时候你需要三件套Base URL、API Key、Model ID或对应的服务标识。以 TaoToken 这类统一通道为例接入信息这样配Base URLhttps://taotoken.net/apiAPI Key在控制台生成形如sk-xxxx服务标识按你订阅的通道填写生成 Key 的入口在控制台的 API Keys 页面文档在接入文档里能查到完整的请求格式。如果你后面要做长期的批量导出任务可以考虑 Coding Plan 这类长期方案避免频繁换 Key。注意无论直连还是走 API 通道凭据都不要写进代码仓库。用环境变量或密钥管理服务这是底线。环境铺好后先跑一个最小连通性测试连上库执行SELECT 1 FROM DUALOracle或SELECT 1MySQL能返回就说明连接没问题。别跳过这步很多「BLOB 读不出来」的锅其实是连接根本没通。3. 可复制配置从连接片段到 BLOB 转本地文件的完整代码这一节是重头戏给你能直接跑的代码。先给直连版本再给 API 通道版本。3.1 Oracle 直连读取 BLOB 并落地import os import oracledb # 中文环境字符集设置避免乱码 os.environ[NLS_LANG] SIMPLIFIED CHINESE_CHINA.UTF8 DB_CONFIG { user: os.getenv(DB_USER, your_user), password: os.getenv(DB_PASSWORD, your_password), dsn: os.getenv(DB_DSN, 10.0.0.10:1521/orcl), } OUTPUT_DIR rD:\allImages os.makedirs(OUTPUT_DIR, exist_okTrue) def save_blob_to_file(blob_obj, file_path): 把 LOB 对象或 bytes 写成文件分块写避免大字段撑爆内存 with open(file_path, wb) as f: if hasattr(blob_obj, read): # cx_Oracle / oracledb 的 LOB 对象 chunk_size 1024 * 1024 # 1MB 一块 while True: chunk blob_obj.read(chunk_size) if not chunk: break f.write(chunk) else: # 已经是 bytes f.write(blob_obj) def main(): conn oracledb.connect(**DB_CONFIG) cursor conn.cursor() cursor.execute(SELECT USER_ID, AVATAR FROM USER_PROFILE WHERE AVATAR IS NOT NULL) count 0 for user_id, blob in cursor: if blob is None: continue file_path os.path.join(OUTPUT_DIR, f{user_id}.jpg) save_blob_to_file(blob, file_path) count 1 print(f已导出 {file_path}) cursor.close() conn.close() print(f共导出 {count} 个文件) if __name__ __main__: main()几个关键点解释一下。第一os.environ[NLS_LANG]必须在建立连接之前设置设晚了不生效。第二LOB 对象用read(chunk_size)分块读不要一次性read()全部大字段会吃光内存。第三open(..., wb)的b不能少否则二进制会被当文本处理写出来就是坏文件。第四输出目录先makedirs(exist_okTrue)避免目录不存在直接报错。3.2 MySQL 版本对照import pymysql conn pymysql.connect( host10.0.0.20, port3306, useryour_user, passwordyour_password, databaseyour_db, charsetutf8mb4, ) cursor conn.cursor() cursor.execute(SELECT id, avatar FROM user_profile WHERE avatar IS NOT NULL) for row_id, blob in cursor: if blob is None: continue with open(f./out/{row_id}.jpg, wb) as f: f.write(blob) # MySQL 的 BLOB 直接就是 bytes cursor.close() conn.close()MySQL 的 BLOB 取出来直接是 bytes不用 read这点和 Oracle 不同别搞混。3.3 走统一 API 通道的版本如果数据服务通过 HTTP 暴露BLOB 通常以 base64 或二进制流返回。假设接口返回 JSON字段是 base64import base64 import os import requests API_BASE https://taotoken.net/api API_KEY os.getenv(TAOTOKEN_API_KEY) def fetch_and_save(record_id, out_path): resp requests.get( f{API_BASE}/blob/{record_id}, headers{Authorization: fBearer {API_KEY}}, timeout60, ) resp.raise_for_status() data resp.json() raw base64.b64decode(data[content]) with open(out_path, wb) as f: f.write(raw) fetch_and_save(1001, ./out/1001.jpg)提示走 API 通道时超时时间要设够大文件传输慢默认几秒很容易断。建议 60 秒起步必要时用流式下载。配置片段给全了接下来验证。4. 验证请求与成功结果怎么确认文件真的落地且完整代码跑完不代表成功得验证。我一般分三层验证文件存在、大小合理、内容可打开。4.1 基础验证文件是否生成import os def verify_files(output_dir, expected_count): files [f for f in os.listdir(output_dir) if f.endswith(.jpg)] print(f期望 {expected_count} 个实际 {len(files)} 个) empty [f for f in files if os.path.getsize(os.path.join(output_dir, f)) 0] if empty: print(f发现空文件{empty}) return files空文件是最常见的「假成功」——程序没报错但文件是 0 字节。原因通常是 BLOB 本身为空或者 read 逻辑写错。4.2 完整性校验大小与哈希如果数据库里能查到 BLOB 的长度Oracle 用DBMS_LOB.GETLENGTHMySQL 用LENGTH可以对照import hashlib def file_md5(path): h hashlib.md5() with open(path, rb) as f: for chunk in iter(lambda: f.read(8192), b): h.update(chunk) return h.hexdigest() # 对照数据库里的长度 db_length 204800 # 从 DBMS_LOB.GETLENGTH 查出来 local_size os.path.getsize(./out/1001.jpg) assert db_length local_size, f大小不一致库 {db_length}本地 {local_size} print(校验通过)大小一致基本就能确认没截断。要更严格就比对 MD5前提是库里存了哈希值。4.3 内容验证能不能打开图片用 Pillow 打开一次PDF 用 pypdf 读一下页数能正常解析就说明二进制没坏from PIL import Image img Image.open(./out/1001.jpg) img.verify() # 不通过会抛异常 print(f图片格式 {img.format}尺寸 {img.size})实测下来这三层验证走完基本不会漏掉坏文件。如果某张图打不开回去查那条记录的 BLOB 是不是本来就损坏而不是怀疑代码。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错对照排错环节按真实报错来遇到哪个查哪个。5.1 401 Unauthorized走 API 通道时最常见。原因就三类Key 没带、Key 写错、Key 过期。检查请求头是不是Authorization: Bearer sk-xxx注意 Bearer 后面有个空格。Key 从控制台重新复制别手打。如果用的是环境变量确认os.getenv真的取到了值打印一下长度看看。5.2 local proxy failed这个报错通常出现在网络请求环节意思是本地网络配置有问题。先确认你的机器能正常访问目标地址用curl -v或requests打印详细错误。如果是公司内网检查是否需要走内部网关。别去折腾系统级网络设置先定位是 DNS 还是端口不通。5.3 reading choices 相关报错如果你在调用模型服务时看到reading choices之类的错误多半是返回体结构和预期不符——比如返回的是错误 JSON代码却按成功结构去取choices[0]。加一层判断data resp.json() if choices not in data: raise RuntimeError(f接口返回异常{data}) content data[choices][0][message][content]先看原始返回再改解析逻辑别盲目重试。5.4 OAuth 相关报错OAuth 报错一般是 token 过期或 scope 不足。刷新 token 后重试如果还不行检查应用申请的权限范围是否包含你要访问的数据接口。这类问题看日志里的error_description字段比错误码有用得多。5.5 BLOB 专属坑类型错误与乱码TypeError: a bytes-like object is required—— 说明你拿到的是 LOB 对象不是 bytes用read()转一下。中文文件名乱码 —— 确认文件系统编码和 Python 默认编码一致Windows 上用os.path.join拼路径一般没问题别手动拼字符串。文件写出来打不开 —— 九成是open少了b或者写入时做了编码转换。注意排错时先打印原始数据类型print(type(blob))比猜快十倍。6. 语义一致收尾把 BLOB 落地这件事做扎实的几点经验最后聊点实操心得不整虚的。批量导出时别一个连接跑到底。数据量大就分批查比如按 ID 区间或时间范围切每批处理完提交一次避免长事务把库拖垮。文件名用主键命名天然去重别用时间戳重复跑会生成一堆垃圾文件。大字段导出建议加进度日志写到文件里中途断了能知道导到哪续跑不用从头来。我一般每处理 100 条打一行日志包含当前 ID 和累计数量。还有一点导出目录和数据库别放同一块盘。BLOB 动辄几个 G把系统盘写满会导致数据库异常这个坑我踩过。如果你是通过统一 API 通道做这件事记得把 Key 的权限收窄到只读别用管理员 Key 跑导出脚本。需要生成 Key 就去控制台的 API Keys 页面接入细节看接入文档长期批量任务可以了解下 Coding Plan。想先验证返回结构对不对用模型对话快速试一下请求格式比直接写脚本调试快。代码能跑通、文件能打开、大小对得上这件事就算落地了。剩下的就是把它包成定时任务或命令行工具交给运维去跑。