我有一个带有多对多用户的 PostgreSQL 数据库，以标记与下表的关系：
- social_user：用户信息
- 标签：标签信息
- user_tag：social_user和标签之间的多对多关系
我正在尝试使用 Flask、Peewee 和 Marshmallow 构建一个简单的 API 来访问该数据库中的数据。我们现在可以忽略 Flask，但我正在尝试为social_user创建一个模式，它允许我转储一个查询，该查询返回一个或多个用户及其各自的标签。我正在寻找类似于以下内容的响应：
{ "id": "[ID]", "handle": "[HANDLE]", "local_id": "[LOCAL_ID]", "platform_slug": "[PLATFORM_SLUG]", "tags": [ { "id": "[ID]", "title": "[TITLE]", "tag_type": "[TAG_TYPE]" }, { "id": "[ID]", "title": "[TITLE]", "tag_type": "[TAG_TYPE]" } ] }
我已经设法通过在 @post_dump 包装函数中包含第二个查询来拉取social_user模式中的标签来做到这一点，但是，这感觉像是一种黑客行为，而且对于大量用户来说似乎会很慢（更新：这很慢，我在 369 个用户上测试过）。我想我可以用 Marshmallow 的fields.Nestedfield type做一些事情。有没有更好的方法可以只用一个 Peewee 查询来序列化这种关系？我的代码如下：
# just so you are aware of my namespaces
import marshmallow as marsh
import peewee as pw
皮威模型
db = postgres_ext.PostgresqlExtDatabase(
register_hstore = False,
**json.load(open('postgres.json'))
)
class Base_Model(pw.Model):
class Meta:
database = db
class Tag(Base_Model):
title = pw.CharField()
tag_type = pw.CharField(db_column = 'type')
class Meta:
db_table = 'tag'
class Social_User(Base_Model):
handle = pw.CharField(null = True)
local_id = pw.CharField()
platform_slug = pw.CharField()
class Meta:
db_table = 'social_user'
class User_Tag(Base_Model):
social_user_id = pw.ForeignKeyField(Social_User)
tag_id = pw.ForeignKeyField(Tag)
class Meta:
primary_key = pw.CompositeKey('social_user_id', 'tag_id')
db_table = 'user_tag'
棉花糖模式
class Tag_Schema(marsh.Schema):
id = marsh.fields.Int(dump_only = True)
title = marsh.fields.Str(required = True)
tag_type = marsh.fields.Str(required = True, default = 'descriptive')
class Social_User_Schema(marsh.Schema):
id = marsh.fields.Int(dump_only = True)
local_id = marsh.fields.Str(required = True)
handle = marsh.fields.Str()
platform_slug = marsh.fields.Str(required = True)
tags = marsh.fields.Nested(Tag_Schema, many = True, dump_only = True)
def _get_tags(self, user_id):
query = Tag.select().join(User_Tag).where(User_Tag.social_user_id == user_id)
tags, errors = tags_schema.dump(query)
return tags
@marsh.post_dump(pass_many = True)
def post_dump(self, data, many):
if many:
for datum in data:
datum['tags'] = self._get_tags(datum['id']) if datum['id'] else []
else:
data['tags'] = self._get_tags(data['id'])
return data
user_schema = Social_User_Schema()
users_schema = Social_User_Schema(many = True)
tags_schema = Tag_Schema(many = True)
以下是一些演示功能的测试：
db.connect()
query = Social_User.get(Social_User.id == 825)
result, errors = user_schema.dump(query)
db.close()
pprint(result)
{'handle': 'test', 'id': 825, 'local_id': 'test', 'platform_slug': 'tw', 'tags': [{'id': 20, 'tag_type': 'descriptive', 'title': 'this'}, {'id': 21, 'tag_type': 'descriptive', 'title': 'that'}]}
db.connect()
query = Social_User.select().where(Social_User.platform_slug == 'tw')
result, errors = users_schema.dump(query)
db.close()
pprint(result)
[{'handle': 'test', 'id': 825, 'local_id': 'test', 'platform_slug': 'tw', 'tags': [{'id': 20, 'tag_type': 'descriptive', 'title': 'this'}, {'id': 21, 'tag_type': 'descriptive', 'title': 'that'}]}, {'handle': 'test2', 'id': 826, 'local_id': 'test2', 'platform_slug': 'tw', 'tags': []}]