如何将 SqlAlchemy 连接查询序列化为 JSON？

Question

谁能帮我设计/找到一个 JSON 编码器，它可以在我的 sqlalchemy 查询中使用连接时工作，并将序列化我指定的结果。

查询如下：data = db.session.query(Post).join(Users).filter(Post.area == id, Users.id == Post.user_id).all()

然后我像这样进行 json 转储：json_object = json.dumps(data, cls=AlchemyEncoder)

传统上我使用自己的 JSON 编码器，可以在这里看到：

class AlchemyEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o.__class__, DeclarativeMeta):
            data = {}
            fields = o.__json__() if hasattr(o, '__json__') else dir(o)
            for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class']]:
                value = o.__getattribute__(field)
                try:
                    json.dumps(value)
                    data[field] = value
                except TypeError:
                    data[field] = None
            return data
        return json.JSONEncoder.default(self, o)

它引用的函数如下所示：

 def __json__(self):
        return ['id', 'user_id', 'message', 'media', 'time_uploaded', 'lifetime', 'votes', 'area']

但是，我希望在我创建的 JSON 对象中加入查询的表中有列。我的 JSON 编码器显然只能使用在我查询的第一个表中找到的列。

编辑：

这是我的两个 SQLAlchemy 类的样子：

class Post(Base):
    __tablename__ = "post"

    id = db.Column('id', UUIDType(binary=False), primary_key=True)
    user_id = db.Column('user_id', UUIDType(binary=False), db.ForeignKey('users.id'))
    votes = db.Column('votes', db.Integer)
    message = db.Column('message', db.Unicode)
    media = db.Column('media', db.Unicode)
    lifetime = db.Column('lifetime', db.Integer)
    time_uploaded = db.Column('time_uploaded', db.DateTime, server_default=func.now())
    area = db.Column('area', db.Unicode)

    user = db.relationship('Users', foreign_keys=user_id)

    def __json__(self):
        return ['id', 'username', 'name', 'profile_picture', 'message', 'media', 'time_uploaded', 'lifetime', 'votes', 'area']


class Users(Base):
    __tablename__ = "users"

    id = db.Column('id', UUIDType(binary=False), primary_key=True)
    username = db.Column('username', db.Unicode, unique=True)
    email = db.Column('email', db.Unicode, unique=True)
    password = db.Column('password', db.Unicode)
    name = db.Column('name', db.Unicode)
    bio = db.Column('bio', db.Unicode)
    profile_picture = db.Column('profile_picture', db.Unicode) 

Answer 1

sqlalchemy 将您传递给query 方法的任何内容放入查询中的select 部分，因此如果您想获得2 个类，您可以这样做：

db.session.query(Post, Users).join(Users).filter(Post.area == id, Users.id == Post.user_id).all()

这将返回 Post 和 Users 类的元组数组（因此您必须更改编码器才能接收元组）

编辑：

这是一个示例，说明如何更改编码器以接受元组：

class AlchemyEncoder(json.JSONEncoder):
    def default(self, o):
        if isinstance(o, tuple):
            data = {}
            for obj in o:
                data.update(self.parse_sqlalchemy_object(obj))
            return data
        if isinstance(o.__class__, DeclarativeMeta):
            return self.parse_sqlalchemy_object(o)
        return json.JSONEncoder.default(self, o)

    def parse_sqlalchemy_object(self, o):
        data = {}
        fields = o.__json__() if hasattr(o, '__json__') else dir(o)
        for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class']]:
            value = o.__getattribute__(field)
            try:
                json.dumps(value)
                data[field] = value
            except TypeError:
                data[field] = None
        return data

Answer 2

现在它适用于 Row，而不是元组。

所以我重写了。

class AlchemyEncoder(json.JSONEncoder):
def default(self, o):
    if isinstance(o, Row):
        data = {}
        for obj in o:
            data.update(self.parse_sqlalchemy_object(obj))
        return data
    if isinstance(o.__class__, DeclarativeMeta):
        return self.parse_sqlalchemy_object(o)
    return json.JSONEncoder.default(self, o)

def parse_sqlalchemy_object(self, o):
    data = {}
    fields = o.__json__() if hasattr(o, '__json__') else dir(o)
    for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class', 'registry']]:
        value = o.__getattribute__(field)
        try:
            json.dumps(value)
            data[field] = value
        except TypeError:
            data[field] = None
    return data


    return json.JSONEncoder.default(self, obj)

谢谢