如何在SQLAlchemy数据库中正确存储与查询numpy数组？

你的问题核心在于numpy数组不能直接以原始形式存入LargeBinary字段——你现在存进去的其实是数组对象的内部二进制表示，而非能直接恢复的序列化数据。下面是几种通用的实现方案，从基础手动处理到优雅的自定义类型都有：

一、基础方案：手动序列化+存储元数据 #

这种方式最直观，适合快速上手，核心是把numpy数组转换成可存储的字节流，同时记录数组的形状（shape）和数据类型（dtype），这样检索时才能完整恢复数组。

1. 修改表结构 #

首先需要在Velocities表中添加两个字段，用来保存数组的元信息：

import sqlalchemy as sql
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Velocities(Base):
    __tablename__ = 'velocities'
    id = sql.Column(sql.Integer, primary_key=True)
    obs_id = sql.Column(sql.Integer, sql.ForeignKey('observations.id'))
    # 存储数组的字节流
    vx_bytes = sql.Column(sql.LargeBinary, nullable=False)
    # 存储数组形状（用字符串存储，比如"500,1000"）
    vx_shape = sql.Column(sql.String, nullable=False)
    # 存储数组数据类型（比如"float64"）
    vx_dtype = sql.Column(sql.String, nullable=False)
    intEn = sql.Column(sql.Float, nullable=False)

2. 插入数据时序列化 #

把numpy数组转换成字节流，同时记录shape和dtype：

import numpy as np

# 假设velocities['vx']是你的numpy数组
vx_array = velocities['vx']
# 转成字节流
vx_bytes = vx_array.tobytes()
# 把shape转成字符串（方便存储）
vx_shape = ",".join(map(str, vx_array.shape))
# 记录dtype名称
vx_dtype = str(vx_array.dtype)

session = DBSession()
new_velocity = Velocities(
    vx_bytes=vx_bytes,
    vx_shape=vx_shape,
    vx_dtype=vx_dtype,
    intEn=int_En,
    obs_id=your_obs_id
)
session.add(new_velocity)
session.commit()

3. 检索数据时反序列化 #

取出字节流后，结合元信息恢复成numpy数组：

a = []
for instance in session.query(Velocities).order_by(Velocities.id):
    # 解析shape
    vx_shape = tuple(map(int, instance.vx_shape.split(",")))
    # 解析dtype
    vx_dtype = np.dtype(instance.vx_dtype)
    # 恢复数组
    vx_array = np.frombuffer(instance.vx_bytes, dtype=vx_dtype).reshape(vx_shape)
    a.append(vx_array)

二、优雅方案：自定义SQLAlchemy类型 #

如果不想每次手动处理序列化/反序列化，可以自定义一个SQLAlchemy类型，让它自动处理numpy数组的存储和检索，使用起来和普通字段一样。

1. 定义自定义类型 #

import sqlalchemy as sql
from sqlalchemy.types import TypeDecorator, LargeBinary
import numpy as np

class NumpyArray(TypeDecorator):
    impl = LargeBinary

    def process_bind_param(self, value, dialect):
        # 存储时：把numpy数组转成包含元数据的字节流
        if value is None:
            return None
        # 把shape、dtype和数组字节一起序列化（这里用pickle，也可以用其他方式）
        import pickle
        return pickle.dumps((value.shape, value.dtype, value.tobytes()))

    def process_result_value(self, value, dialect):
        # 检索时：从字节流恢复numpy数组
        if value is None:
            return None
        import pickle
        shape, dtype, bytes_data = pickle.loads(value)
        return np.frombuffer(bytes_data, dtype=dtype).reshape(shape)

2. 使用自定义类型 #

修改表结构，直接用这个类型定义字段：

class Velocities(Base):
    __tablename__ = 'velocities'
    id = sql.Column(sql.Integer, primary_key=True)
    obs_id = sql.Column(sql.Integer, sql.ForeignKey('observations.id'))
    # 直接用自定义类型
    vx = sql.Column(NumpyArray, nullable=False)
    intEn = sql.Column(sql.Float, nullable=False)

3. 插入和检索数据 #

这时候就和操作普通字段完全一样了，不需要手动处理：

# 插入
session = DBSession()
new_velocity = Velocities(
    vx=velocities['vx'],  # 直接传numpy数组
    intEn=int_En,
    obs_id=your_obs_id
)
session.add(new_velocity)
session.commit()

# 检索
a = []
for instance in session.query(Velocities).order_by(Velocities.id):
    # instance.vx直接就是numpy数组，可以直接使用/绘图
    a.append(instance.vx)

三、额外注意事项 #

• 关于pickle的安全性：上面的自定义类型用了pickle序列化，如果你数据库中的数据可能来自不可信来源，pickle存在安全风险，建议改用更安全的序列化方式（比如结合tobytes()+手动存储shape/dtype的方式，或者用JSON存储元数据+字节流）。
• 压缩优化：如果你的数组很大（比如500x1000的数组），可以考虑在存储前用zlib压缩字节流，减少存储空间：

  import zlib
  # 存储时压缩
  compressed_bytes = zlib.compress(vx_array.tobytes())
  # 检索时解压
  bytes_data = zlib.decompress(compressed_bytes)
  

• 性能考量：大数组的存储和检索会影响数据库性能，如果频繁操作这类数据，建议考虑专门的数值型数据库（比如PostgreSQL的array类型配合numpy，但这属于特定数据库方案，不符合你要的通用需求），或者把数组存储在文件系统，数据库只存文件路径。

内容的提问来源于stack exchange，提问作者Chicrala