烧瓶:启动时插入一些测试数据?

当flask第一次启动时,我正在尝试将一些虚拟数据插入到我的数据库中。我正在用 flask-migrate

我跑:

• flask db init
• flask db migrate
• flask db upgrade

我检查了迁移脚本,一切看起来都很好。

我甚至跑了 insert into keywords (keyword) values ('test'); 在Dbeaver中只是为了验证我可以插入一些数据。

flask run ,我得到一个 sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) no such table: keywords ??

回溯是错误的根源 existing = Keyword.query.filter(Keyword.keyword == k.lower()).one_or_none() create_app 功能。

整个 创建应用程序 函数如下所示:

def create_app(config='metadata.settings'):
    app = Flask(__name__, instance_relative_config=True)

    app.config.from_object(config)
    load_extensions(app)  # sets db.init_app(app)

    @app.before_first_request
    def insert_test_data():
        test_record = {
            "metadata_version": "1.1",
            "name": "foo-package",
            "version": "0.1",
            "platforms": [
                "UNKNOWN"
            ],
            "summary": "Hello Foo Package",
            "keywords": "foo pandas numpy",
            "home_page": "https://gitlab.blah.com/my-examples/model-bar",
            "author": "John Smith",
            "author_email": "john.smith@gmail.com",
            "classifiers": [
                "Programming Language :: Python :: 3",
                "Operating System :: Microsoft :: Windows :: Windows 10",
                "Development Status :: 1 - Planning",
                "Topic :: Software Development :: Testing :: Mocking",
                "Topic :: Software Development :: Testing"
            ]
        }

        def add_record(test_record):
            rec = PackageMetadata()
            rec.name = test_record['name'],
            rec.version = test_record['version'],
            rec.desc = test_record['summary'],
            rec.home_page = test_record['home_page'],
            rec.author = test_record['author'],
            rec.author_email = test_record['author_email']

            keywords = test_record['keywords'].split(' ')
            for keyword in keywords:
                keyword_to_add = add_keyword(keyword)
                rec.keywords.append(keyword_to_add)

            classifiers = test_record['classifiers']
            for classifier in classifiers:
                classifier_to_add = add_classifier(classifier)
                rec.classifiers.append(classifier_to_add)

            try:
                db.session.add(rec)
                db.session.commit()
            except Exception as e:
                print(e)

        def add_keyword(k):
            existing = Keyword.query.filter(Keyword.keyword == k.lower()).one_or_none()
            if existing is not None:
                return existing
            else:
                new_keyword = Keyword()
                new_keyword.keyword = k.lower()
                return new_keyword

        def add_classifier(k):
            existing = Classifier.query.filter(Classifier.classifier == k).one_or_none()
            if existing is not None:
                return existing
            else:
                new_classifier = Classifier()
                new_classifier.classifier = k
                return new_classifier
        add_record(test_record)
    return app