Flask-Restx 未将枚举字段类型转换为 JSON
Flask-Restx not converting enum field type to JSON
我需要有关枚举字段类型的帮助,因为它不被 Swagger 接受并且我收到错误消息 **TypeError: Object or Type eGameLevel is not JSON serializable**
。下面是 table 的完整代码集。提供了带有 DB table 和 sqlalchemy 设置的完整代码集。我已经用 Marshmallow-Enum Flask 包试过了,但没有用。寻求有关解决方案的一些解释的帮助,以便我可以很好地学习它。 :-)
我正在使用 MySQL 和 Flask。在 Postgres 中,管理所有选择字段非常容易。我只需要一个工作示例或 link 到存储库,其中 MySQL 选择字段显示在 swagger 下拉列表中。
我的模特:
import enum
from app import db
from typing import List
class eGameLevel(enum.Enum):
BEGINNER = 'Beginner'
ADVANCED = 'Advanced'
class Game(Base):
__tablename__ = 'game_stage'
id = db.Column(db.Integer(), primary_key=True)
game_level= db.Column(db.Enum(eGameLevel),
default=eGameLevel.BEGINNER, nullable=False)
user_id = db.Column(db.Integer(), db.ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
user = db.relationship('User', backref='game__level_submissions', lazy=True)
def __init__(self, game_level, user_id):
self.game_level = game_level
self.user_id = user_id
def __repr__(self):
return 'Game(game_level%s, ' \
'user_id%s'%(self.game_level,
self.user_id)
def json(self):
return {'game_level':self.game_level,
'user_id':self.user_id}
@classmethod
def by_game_id(cls, _id):
return cls.query.filter_by(id=_id)
@classmethod
def find_by_game_level(cls, game_level):
return cls.query.filter_by(game_level=game_level)
@classmethod
def by_user_id(cls, _user_id):
return cls.query.filter_by(user_id=_user_id)
@classmethod
def find_all(cls) -> List["Game"]:
return cls.query.all()
def save_to_db(self) -> None:
db.session.add(self)
db.session.commit()
def delete_from_db(self) -> None:
db.session.delete(self)
db.session.commit()
我的架构
from app import ma
from app.models import Gode
class GameSchema(ma.SQLAlchemyAutoSchema):
game = ma.Nested('GameSchema', many=True)
class Meta:
model = Game
load_instance = True
include_fk= True
我的资源:
from flask_restx import Resource, fields, Namespace
from app.models import Game
from app import db
from app.schemas import GameSchema
GAME_REQUEST_NOT_FOUND = "Game request not found."
GAME_REQUEST_ALREADY_EXSISTS = "Game request '{}' Already exists."
game_ns = Namespace('Game', description='Available Game Requests')
games_ns = Namespace('Game Requests', description='All Games Requests')
game_schema = GameSchema()
games_list_schema = GameSchema(many=True)
gamerequest = game_ns.model('Game', {
'game_level': fields.String('Game Level: Must be one of: BEGINNER, ADVANCED.'),
'user_id': fields.Integer,
})
class GameRequestsListAPI(Resource):
@games_ns.doc('Get all Game requests.')
def get(self):
return games_list_schema.dump(Game.find_all()), 200
@games_ns.expect(gamerequest)
@games_ns.doc("Create a Game request.")
def post(self):
game_json = request.get_json()
game_data = game_schema.load(game_json)
game_data.save_to_db()
return game_schema.dump(game_data), 201
而不是使用枚举:
class eGameLevel(enum.Enum):
BEGINNER = 'Beginner'
ADVANCED = 'Advanced'
你可以使用字典:
eGameLevel = {"BEGINNER": 1, "ADVANCED": 2}
然后您可以将 sql 数据模型的枚举类型替换为字符串类型,如:
game_level= db.Column(db.Integer(),
default=eGameLevel["BEGINNER"], nullable=False)
并在整个应用程序中使用定义的字典进行适当的检查。这也将解决 alembic 以及进行数据库迁移的问题。
您还需要修改一些 python 文件。我宁愿直接在这里做,然后你可以查找修改它们:
#Import at Resources
from flask import request
from app.models import Game, eGameLevel
Post 部分:
# For post part
payload = request.json
game_obj = Game(game_level=eGameLevel[payload["game_level"]], user_id=payload["user_id"])
db.session.add(game_obj)
db.session.commit()
另外,我没看懂from app.models import Gode
是什么意思。
我无法 运行 你的例子,似乎有些文件丢失了,你的项目结构我也不清楚。您可以共享一个包含所有源代码和需求文件的存储库吗?我做了一个小例子来测试 Enum
的序列化
from enum import Enum
import sqlalchemy as sa
from flask import Flask
from flask_restx import Api, Namespace, Resource
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
class eGameLevel(str, Enum):
BEGINNER = "Beginner"
ADVANCED = "Advanced"
engine = sa.create_engine("sqlite:///:memory:")
session = scoped_session(sessionmaker(bind=engine))
Base = declarative_base()
class Game(Base):
__tablename__ = "game"
id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
level = sa.Column(sa.Enum(eGameLevel), default=eGameLevel.BEGINNER, nullable=False)
def __repr__(self):
return f"Game(id={self.id}, level={self.level})"
def json(self):
data = {"id": self.id, "level": self.level}
return data
Base.metadata.create_all(engine)
g1 = Game(level=eGameLevel.BEGINNER)
g2 = Game(level=eGameLevel.ADVANCED)
session.add_all([g1, g2])
session.commit()
query_content = session.query(Game).all()
games_ns = Namespace("Game Requests", description="All Games Requests")
app = Flask(__name__)
api = Api(app)
@api.route("/game")
class GameRequestsListAPI(Resource):
@games_ns.doc("Get all Game requests.")
def get(self):
data = [x.json() for x in query_content]
return data, 200
app.run(debug=True)
这个例子有效,我认为由于枚举声明中的 str
序列化是可能的:class eGameLevel(str, Enum)
.
如果你分享整个源代码,我可以尝试修复它。
与其尝试管理 MySQL 架构的枚举字段,我建议使用另一个 table 和对您的 eGameLevel 的反向引用。您可以摆脱这整个大惊小怪,而且将来如果您需要在您的选择字段中添加另一个选项,您将不必对其进行硬编码。
只需创建一个 main table 作为 Game 和 sub table 作为 eGameLevel(只有一个字符串字段)。您将能够从您的游戏 table.
中访问选择
每当我遇到困难时,我都会按照 here 中提到的基础知识进行学习。
我需要有关枚举字段类型的帮助,因为它不被 Swagger 接受并且我收到错误消息 **TypeError: Object or Type eGameLevel is not JSON serializable**
。下面是 table 的完整代码集。提供了带有 DB table 和 sqlalchemy 设置的完整代码集。我已经用 Marshmallow-Enum Flask 包试过了,但没有用。寻求有关解决方案的一些解释的帮助,以便我可以很好地学习它。 :-)
我正在使用 MySQL 和 Flask。在 Postgres 中,管理所有选择字段非常容易。我只需要一个工作示例或 link 到存储库,其中 MySQL 选择字段显示在 swagger 下拉列表中。
我的模特:
import enum
from app import db
from typing import List
class eGameLevel(enum.Enum):
BEGINNER = 'Beginner'
ADVANCED = 'Advanced'
class Game(Base):
__tablename__ = 'game_stage'
id = db.Column(db.Integer(), primary_key=True)
game_level= db.Column(db.Enum(eGameLevel),
default=eGameLevel.BEGINNER, nullable=False)
user_id = db.Column(db.Integer(), db.ForeignKey('users.id', ondelete='CASCADE'), nullable=False)
user = db.relationship('User', backref='game__level_submissions', lazy=True)
def __init__(self, game_level, user_id):
self.game_level = game_level
self.user_id = user_id
def __repr__(self):
return 'Game(game_level%s, ' \
'user_id%s'%(self.game_level,
self.user_id)
def json(self):
return {'game_level':self.game_level,
'user_id':self.user_id}
@classmethod
def by_game_id(cls, _id):
return cls.query.filter_by(id=_id)
@classmethod
def find_by_game_level(cls, game_level):
return cls.query.filter_by(game_level=game_level)
@classmethod
def by_user_id(cls, _user_id):
return cls.query.filter_by(user_id=_user_id)
@classmethod
def find_all(cls) -> List["Game"]:
return cls.query.all()
def save_to_db(self) -> None:
db.session.add(self)
db.session.commit()
def delete_from_db(self) -> None:
db.session.delete(self)
db.session.commit()
我的架构
from app import ma
from app.models import Gode
class GameSchema(ma.SQLAlchemyAutoSchema):
game = ma.Nested('GameSchema', many=True)
class Meta:
model = Game
load_instance = True
include_fk= True
我的资源:
from flask_restx import Resource, fields, Namespace
from app.models import Game
from app import db
from app.schemas import GameSchema
GAME_REQUEST_NOT_FOUND = "Game request not found."
GAME_REQUEST_ALREADY_EXSISTS = "Game request '{}' Already exists."
game_ns = Namespace('Game', description='Available Game Requests')
games_ns = Namespace('Game Requests', description='All Games Requests')
game_schema = GameSchema()
games_list_schema = GameSchema(many=True)
gamerequest = game_ns.model('Game', {
'game_level': fields.String('Game Level: Must be one of: BEGINNER, ADVANCED.'),
'user_id': fields.Integer,
})
class GameRequestsListAPI(Resource):
@games_ns.doc('Get all Game requests.')
def get(self):
return games_list_schema.dump(Game.find_all()), 200
@games_ns.expect(gamerequest)
@games_ns.doc("Create a Game request.")
def post(self):
game_json = request.get_json()
game_data = game_schema.load(game_json)
game_data.save_to_db()
return game_schema.dump(game_data), 201
而不是使用枚举:
class eGameLevel(enum.Enum):
BEGINNER = 'Beginner'
ADVANCED = 'Advanced'
你可以使用字典:
eGameLevel = {"BEGINNER": 1, "ADVANCED": 2}
然后您可以将 sql 数据模型的枚举类型替换为字符串类型,如:
game_level= db.Column(db.Integer(),
default=eGameLevel["BEGINNER"], nullable=False)
并在整个应用程序中使用定义的字典进行适当的检查。这也将解决 alembic 以及进行数据库迁移的问题。
您还需要修改一些 python 文件。我宁愿直接在这里做,然后你可以查找修改它们:
#Import at Resources
from flask import request
from app.models import Game, eGameLevel
Post 部分:
# For post part
payload = request.json
game_obj = Game(game_level=eGameLevel[payload["game_level"]], user_id=payload["user_id"])
db.session.add(game_obj)
db.session.commit()
另外,我没看懂from app.models import Gode
是什么意思。
我无法 运行 你的例子,似乎有些文件丢失了,你的项目结构我也不清楚。您可以共享一个包含所有源代码和需求文件的存储库吗?我做了一个小例子来测试 Enum
的序列化from enum import Enum
import sqlalchemy as sa
from flask import Flask
from flask_restx import Api, Namespace, Resource
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker
class eGameLevel(str, Enum):
BEGINNER = "Beginner"
ADVANCED = "Advanced"
engine = sa.create_engine("sqlite:///:memory:")
session = scoped_session(sessionmaker(bind=engine))
Base = declarative_base()
class Game(Base):
__tablename__ = "game"
id = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
level = sa.Column(sa.Enum(eGameLevel), default=eGameLevel.BEGINNER, nullable=False)
def __repr__(self):
return f"Game(id={self.id}, level={self.level})"
def json(self):
data = {"id": self.id, "level": self.level}
return data
Base.metadata.create_all(engine)
g1 = Game(level=eGameLevel.BEGINNER)
g2 = Game(level=eGameLevel.ADVANCED)
session.add_all([g1, g2])
session.commit()
query_content = session.query(Game).all()
games_ns = Namespace("Game Requests", description="All Games Requests")
app = Flask(__name__)
api = Api(app)
@api.route("/game")
class GameRequestsListAPI(Resource):
@games_ns.doc("Get all Game requests.")
def get(self):
data = [x.json() for x in query_content]
return data, 200
app.run(debug=True)
这个例子有效,我认为由于枚举声明中的 str
序列化是可能的:class eGameLevel(str, Enum)
.
如果你分享整个源代码,我可以尝试修复它。
与其尝试管理 MySQL 架构的枚举字段,我建议使用另一个 table 和对您的 eGameLevel 的反向引用。您可以摆脱这整个大惊小怪,而且将来如果您需要在您的选择字段中添加另一个选项,您将不必对其进行硬编码。
只需创建一个 main table 作为 Game 和 sub table 作为 eGameLevel(只有一个字符串字段)。您将能够从您的游戏 table.
中访问选择每当我遇到困难时,我都会按照 here 中提到的基础知识进行学习。