如何在 sqlalchemy 的 with_entities() 中使用 python 代码？

Question

我的代码是这样的：

status_dict = {
    0: 'ready',
    1: 'downloading',
    2: 'downloaded',
}


class Videos(db.Model):
    __tablename__ = 'videos'
    id = db.Column(db.Integer, primary_key=True)
    status = db.Column(db.Integer, default=0, index=True)


class Test:

    @classmethod
    def run(cls):
        videos = Videos.query.with_entities(
            Videos.id.label('vid'),
            Videos.status,
            some_func(status_dict[Videos.status]).label('status_desc')
        ).order_by(
            Videos.id.desc()
        ).all()

        rows = qs2json(videos)

我想达到这样的效果：

[
    {
        "vid": 103,
        "status": 0,
        "status_desc": "ready",
    },
    {
        "vid": 102,
        "status": 1,
        "status_desc": "downloading",
    },
    {
        "vid": 101,
        "status": 2,
        "status_desc": "downloaded",
    }
]

我已经实现了函数qs2json()，但是SQLAlchemy中的函数some_func是什么？

我在 SQLAlchemy 中尝试了一些函数，但遇到了错误

---

我做了一个这样的自定义python函数：

from sqlalchemy.sql.functions import GenericFunction, func
from sqlalchemy.types import String


class MyFuncs(GenericFunction):
    type = String

    def get_status_desc(self, status):
        return status_dict[status]


class Test:

    @classmethod
    def run(cls):
        videos = Videos.query.with_entities(
            Videos.id.label('vid'),
            Videos.status,
            func.MyFuncs.get_status_desc(Videos.status).label('status_desc')
        ).order_by(
            Videos.id.desc()
        ).all()

我遇到了sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1305, 'FUNCTION MyFuncs.get_status_desc does not exist')

Answer 1

为此使用case，作为您的示例：

import sqlalchemy as sa
class Videos(db.Model):
    __tablename__ = 'videos'
    id = db.Column(db.Integer, primary_key=True)
    status = db.Column(db.Integer, default=0, index=True)

    @classmethod
    def as_dict(cls, obj):
        return dict(obj)


videos = Videos.query.with_entities(
    Videos.id.label('vid'),
    Videos.status,
    sa.case([(Videos.status == k, v) for k, v in status_dict.items()]).label('status_desc')
).order_by(
    Videos.id.desc()
).all()

json_data = json.dumps([Videos.as_dict(v) for v in videos])

print(json_data)
# [{"vid": 3, "status": 1, "status_desc": "downloading"}, {"vid": 2, "status": 2, "status_desc": "downloaded"}, {"vid":
# 1, "status": 0, "status_desc": "ready"}]