我正在使用应用程序工厂模式,并且我有一个与 M2M 关系中的类别相关的文章对象。到目前为止,API 中的所有路由都按预期工作。我可以通过 POST MethodViews 创建带有类别的文章。
但是,我正在尝试通过单击单独的文件将一些示例数据播种到我的数据库中。起初我认为问题出在 Flask CLI 和应用程序上下文,我最初在蓝图中有,但后来我意识到问题更深了一些。我看到了这个问题,但我已将我的 Marshmallow/Flask-Marshmallow/Marshmallow-Sqlalchemy 更新到最新版本:
https://github.com/marshmallow-code/marshmallow-sqlalchemy/issues/20#issuecomment-136400602
// 模型.py
class CRUDMixin():
@classmethod
def find_by_id(cls, _id):
return cls.query.filter_by(id=_id).first()
@classmethod
def find_all(cls):
return cls.query.all()
def save_to_db(self):
db.session.add(self)
return db.session.commit()
def update(self):
return db.session.commit()
def delete_from_db(self):
db.session.delete(self)
return db.session.commit()
class CategoryModel(db.Model, CRUDMixin):
__tablename__ = "api_category"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), nullable=False)
def __repr__(self):
return '<id {}>'.format(self.id)
class ArticleModel(db.Model, CRUDMixin):
__tablename__ = 'api_article'
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), nullable=False)
description = db.Column(db.Text)
categories = db.relationship('CategoryModel', secondary=api_category_article, lazy='subquery',
backref=db.backref('articles', lazy=True))
def __repr__(self):
return '<id {}>'.format(self.id)
// 架构.py
class CategoryPostSchema(ma.SQLAlchemyAutoSchema):
class Meta:
model = CategoryModel
dump_only = ("name", )
load_only = ("articles", )
load_instance = True
class ArticlePostSchema(ma.SQLAlchemyAutoSchema):
categories = ma.Nested(CategoryPostSchema, many=True)
class Meta:
model = ArticleModel
dump_only = ("id",)
include_fk = True
load_instance = True
include_relationships = True
sqla_session = db.session
// 资源.py
class ArticleListView(MethodView):
def __init__(self):
pass
@classmethod
def get(cls):
data = ArticleModel.find_all()
return jsonify({"data": article_list_schema.dump(data),
"count": len(data),
"status": 200
})
@classmethod
def post(cls):
req_json = request.get_json()
errors = article_post_schema.validate(req_json)
if errors:
response = jsonify({'errors': errors, "status": 400})
response.status_code = 400
return response
data = article_post_schema.load(req_json)
data.save_to_db()
response = jsonify({"data": article_post_schema.dump(data), "errors": {}, "status": 201})
response.status_code = 201
return response
// initialize.py (在根目录)
import click
from marshmallow import ValidationError
from api import create_app
from api.db import db
from api.schemas import ArticlePostSchema
app = create_app()
@click.group()
def cli():
pass
@cli.command()
def article():
with app.app_context():
article_post_schema = ArticlePostSchema()
entry = {"name":"My Family Vacation 5",
"description":"That time we took a road trip to Europe",
"categories":[{"id": 1}]
}
data = article_post_schema.load(entry, session=db.session)
data.save_to_db()
print("Success")
if __name__ == '__main__':
cli()
// 错误
Traceback (most recent call last):
File "initialize.py", line 41, in <module>
cli()
...
File "initialize.py", line 29, in article
data = article_post_schema.validate(entry)
...
return self.session.query(self.opts.model).filter_by(**filters).first()
AttributeError: 'DummySession' object has no attribute 'query'