使用SQLAlchemy ORM从PostgreSQL取数据遇问题求助

问题背景 #

在使用SQLAlchemy ORM从PostgreSQL表中获取数据时遇到两个问题，需要按条件获取所有行的全量字段数据。

表模型定义 #

class EventMediaFile(db.Model):
    __tablename__ = config["eventmediatableName"]
    EventID = db.Column(
        db.String(120),
        primary_key=True
    )
    CameraID = db.Column(db.String(120), primary_key=True)
    Type = db.Column(db.String(120), primary_key=True)
    URL = db.Column(db.String(120), nullable=True)
    Width = db.Column(db.Integer, nullable=True)
    Height = db.Column(db.Integer, nullable=True)
    MIMEType = db.Column(db.String(120), nullable=True)
    Filesize = db.Column(db.Integer, nullable=True)
    Duration = db.Column(db.Integer, nullable=True)
    FrameCount = db.Column(db.Integer, nullable=True)
    Codec = db.Column(db.String(120), nullable=True)
    Extension = db.Column(db.String(120), nullable=True)
    OriginalFilename = db.Column(db.String(120), nullable=True)
    Status = db.Column(db.String(120), nullable=True)
    LastUpdatedDT = db.Column(db.DateTime(timezone=False), nullable=False)
    GroupID = db.Column(db.String(120), nullable=False)
    DeviceID = db.Column(db.String(120), nullable=False)
    RetryCount = db.Column(db.Integer, nullable=False)
    LastEncounteredError = db.Column(db.String(120), nullable=True)
    ErrorLog = db.Column(db.String(120), nullable=True)

    db.UniqueConstraint(EventID, CameraID, Type)

    def __init__(
        self,
        eventID,
        cameraID,
        type,
        url,
        width,
        height,
        mimeType,
        fileSize,
        duration,
        frameCount,
        codec,
        extension,
        originalFilename,
        status,
        lastUpdatedDT,
        groupID,
        deviceID,
        retryCount,
        lastEncounteredError,
        errorLog
    ):
        self.EventID = eventID
        self.CameraID = cameraID
        self.Type = type
        self.URL = url
        self.Width = width
        self.Height = height
        self.MIMEType = mimeType
        self.Filesize = fileSize
        self.Duration = duration
        self.FrameCount = frameCount
        self.Codec = codec
        self.Extension = extension
        self.OriginalFilename = originalFilename
        self.Status = status
        self.LastUpdatedDT = lastUpdatedDT
        self.GroupID = groupID
        self.DeviceID = deviceID
        self.RetryCount = retryCount
        self.LastEncounteredError = lastEncounteredError
        self.ErrorLog = errorLog

App与DB初始化方式 #

app = Flask(__name__)
db = SQLAlchemy(app)

问题分析与解决 #

问题1：查询仅展示主键，无法获取全量字段 #

原查询代码：

def get_downloadable_files_from_EM(app, db):
    with app.app_context():
        stmt = select(EventMediaFile).where(
            or_(EventMediaFile.Status == 'Missing', EventMediaFile.Status == 'Failed')).where(
            EventMediaFile.RetryCount < 3)
        result = db.session.execute(stmt)
        return result.fetchall()

返回示例：<EventMediaFile event123, camid1, Image>

原因：这是SQLAlchemy模型实例的默认__repr__输出，仅展示主键字段，但实例本身包含所有字段数据。

解决方法：

1. 直接访问实例属性并转字典：遍历结果时，将模型实例转为包含所有字段的字典：

def get_downloadable_files_from_EM(app, db):
    with app.app_context():
        stmt = select(EventMediaFile).where(
            or_(EventMediaFile.Status == 'Missing', EventMediaFile.Status == 'Failed'),
            EventMediaFile.RetryCount < 3
        )
        result = db.session.execute(stmt)
        records = [row[0] for row in result.fetchall()]
        # 自动生成包含所有字段的字典列表
        return [
            {col.name: getattr(record, col.name) for col in EventMediaFile.__table__.columns}
            for record in records
        ]

2. 自定义__repr__方法：在EventMediaFile模型中添加自定义输出，展示更多字段：

def __repr__(self):
    return f"<EventMediaFile EventID:{self.EventID}, CameraID:{self.CameraID}, Status:{self.Status}, URL:{self.URL}>"

问题2：使用db.session.query时提示"Working outside of application context" #

原尝试代码：

result = db.session.query(EventMediaFile).where(
    or_(EventMediaFile.Status == 'Missing', EventMediaFile.Status == 'Failed')).where(
    EventMediaFile.RetryCount < 3)

原因：若返回未执行的Query对象，在应用上下文外执行会触发错误；或传入的db对象与上下文关联异常。

解决方法：在上下文内完成查询并取出数据，不要返回未执行的Query对象：

def get_downloadable_files_from_EM(app, db):
    with app.app_context():
        records = db.session.query(EventMediaFile).filter(
            or_(EventMediaFile.Status == 'Missing', EventMediaFile.Status == 'Failed'),
            EventMediaFile.RetryCount < 3
        ).all()
        # 转为全量字段字典返回
        return [
            {col.name: getattr(rec, col.name) for col in EventMediaFile.__table__.columns}
            for rec in records
        ]

最终可用查询函数 #

整合后的函数，同时解决两个问题：

from sqlalchemy import or_

def get_downloadable_files_from_EM(app, db):
    with app.app_context():
        stmt = select(EventMediaFile).where(
            or_(EventMediaFile.Status == 'Missing', EventMediaFile.Status == 'Failed'),
            EventMediaFile.RetryCount < 3
        )
        result = db.session.execute(stmt)
        records = [row[0] for row in result.fetchall()]
        
        return [
            {col.name: getattr(record, col.name) for col in EventMediaFile.__table__.columns}
            for record in records
        ]

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