如何使用SQLAlchemy保存与检索Blob文件？求清晰实现流程

没问题，我帮你梳理一套一步到位的流程，从模型定义到实际存取都讲得明明白白，用最常用的SQLAlchemy ORM方式来做，你跟着步骤走就行～

1. 先搞定基础环境准备 #

• 先确保装了SQLAlchemy：pip install sqlalchemy
• 如果用非SQLite的数据库（比如MySQL/PostgreSQL），还要装对应驱动：MySQL用mysql-connector-python，PostgreSQL用psycopg2-binary
• 示例就用SQLite了，不用额外装驱动，方便快速测试

2. 定义包含Blob字段的数据库模型 #

首先得创建一个数据库模型，专门用来存Blob文件。SQLAlchemy里用LargeBinary类型对应数据库的Blob字段，咱们再加上几个实用字段方便管理：

from sqlalchemy import create_engine, Column, Integer, String, LargeBinary
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import os
import mimetypes

# 初始化基础类和数据库连接
Base = declarative_base()
# SQLite数据库文件会存在当前目录的file_storage.db里
engine = create_engine('sqlite:///file_storage.db')
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 核心Blob存储模型
class FileStorage(Base):
    __tablename__ = "file_storage"
    
    id = Column(Integer, primary_key=True, index=True)
    filename = Column(String(255), nullable=False)  # 存原文件名，方便后续识别
    content_type = Column(String(100), nullable=False)  # 文件MIME类型，比如image/png
    file_data = Column(LargeBinary, nullable=False)  # 真正存二进制内容的Blob字段

# 自动创建数据库表（第一次运行时执行）
Base.metadata.create_all(bind=engine)

3. 把文件保存到数据库（Blob写入） #

写个实用函数，不管是本地文件还是Web上传的文件，都能轻松存入数据库。分两种场景给你示例：

场景1：保存本地文件到数据库 #

def save_local_file(db_session, file_path, filename=None, content_type=None):
    # 没传文件名就用路径里的原始文件名
    if not filename:
        filename = os.path.basename(file_path)
    # 没传MIME类型就自动识别，识别不出来就用默认二进制类型
    if not content_type:
        content_type, _ = mimetypes.guess_type(file_path)
        content_type = content_type or "application/octet-stream"
    
    # 读取文件二进制内容
    with open(file_path, "rb") as f:
        file_binary = f.read()
    
    # 创建模型实例并提交到数据库
    new_file = FileStorage(
        filename=filename,
        content_type=content_type,
        file_data=file_binary
    )
    db_session.add(new_file)
    db_session.commit()
    return new_file.id  # 返回文件ID，方便后续检索

# 使用示例
db = SessionLocal()
try:
    saved_file_id = save_local_file(db, "test_photo.jpg")
    print(f"文件保存成功！ID是：{saved_file_id}")
finally:
    db.close()  # 用完记得关会话

场景2：Web框架中保存上传的文件（比如FastAPI/Flask） #

以FastAPI的UploadFile为例，直接读取上传文件的二进制内容就行：

from fastapi import UploadFile

def save_uploaded_file(db_session, upload_file: UploadFile):
    file_binary = upload_file.read()
    new_file = FileStorage(
        filename=upload_file.filename,
        content_type=upload_file.content_type,
        file_data=file_binary
    )
    db_session.add(new_file)
    db_session.commit()
    return new_file.id

4. 从数据库读取Blob文件（检索+导出） #

同样写个函数，根据文件ID或者文件名查询，然后导出到本地或者返回给客户端：

场景1：把Blob导出到本地文件 #

def export_blob_to_local(db_session, file_id, save_path=None):
    # 根据ID查询文件记录
    file_record = db_session.query(FileStorage).filter(FileStorage.id == file_id).first()
    if not file_record:
        print("找不到对应的文件！")
        return None
    
    # 没指定保存路径就用原文件名+retrieved前缀
    if not save_path:
        save_path = f"retrieved_{file_record.filename}"
    
    # 把二进制内容写入本地文件
    with open(save_path, "wb") as f:
        f.write(file_record.file_data)
    print(f"文件已导出到：{save_path}")
    return save_path

# 使用示例
db = SessionLocal()
try:
    export_blob_to_local(db, file_id=1)
finally:
    db.close()

场景2：Web框架中返回Blob给客户端（比如FastAPI） #

直接返回带二进制内容的响应，自动让浏览器下载：

from fastapi import Response

def return_blob_to_client(db_session, file_id):
    file_record = db_session.query(FileStorage).filter(FileStorage.id == file_id).first()
    if not file_record:
        return None
    
    return Response(
        content=file_record.file_data,
        media_type=file_record.content_type,
        headers={"Content-Disposition": f"attachment; filename={file_record.filename}"}
    )

5. 必须注意的几个关键点 #

• 大小限制：如果文件超过100MB，不建议存在数据库里，会拖慢数据库性能，这种情况更适合用本地文件系统+存路径到数据库，或者云存储服务
• 会话管理：一定要用try-finally或者上下文管理器（with SessionLocal() as db:）管理数据库会话，避免连接泄漏
• 编码问题：Blob存的是纯二进制，别转成字符串操作，直接读写字节就行，否则会乱码
• 数据库兼容性：不同数据库对Blob的类型命名不同（比如MySQL是LONGBLOB，PostgreSQL是BYTEA），但SQLAlchemy的LargeBinary会自动适配，不用手动改

内容的提问来源于stack exchange，提问作者Jasper Nichol M Fabella