SQLAlchemy 不使用 db.create_all() 创建表

Question

我对 SQLAlchemy 感到非常沮丧。 20 个小时以来，我一直在试图弄清楚为什么 db.create_all() 没有创建任何表。 我还在 StackOverflow 上阅读了一些类似的问题，但没有发现任何对我有帮助的东西。

这是我的代码： another_flask_backend\tests\test_api.py:

import unittest
import json
from datetime import datetime, date

from another_flask_backend.app import create_app
from another_flask_backend.app.models.db import db
from another_flask_backend.config import Config
from another_flask_backend.constants import HttpStatusCodes
from another_flask_backend.app.models.db import User, PersonalityTrait, Interest

class APITestUser(unittest.TestCase):
    def setUp(self):
        """Is called before each test"""
        self.app = create_app(Config.TESTING)
        self.app_context = self.app.app_context()
        self.app_context.push()
        from another_flask_backend.app.models.db import User, PersonalityTrait, Interest
        db.create_all()
        self.client = self.app.test_client()

    def tearDown(self):
        """Is called after each test"""
        db.session.remove()
        db.drop_all()
        self.app_context.pop()

    # lots of unit tests that fails because no tables were created in setUp()

another_flask_backend\app\models\db.py

from typing import TypeVar
from sqlalchemy.orm.state import InstanceState
from flask_sqlalchemy import SQLAlchemy
# coding: utf-8
from sqlalchemy import Column, DECIMAL, DateTime, ForeignKey, String
from sqlalchemy.dialects.mysql import INTEGER, TINYINT
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
db = SQLAlchemy()
Base.query = db.session.query_property()
metadata = Base.metadata


C = TypeVar("C")


def add_repr_method(cls: C) -> C:
    def repr_func(self) -> str:
        return str({k: v for k, v in self.__dict__.items() if not isinstance(v, InstanceState)})

    setattr(cls, '__repr__', repr_func)
    return cls


@add_repr_method
class Interest(Base):
    __tablename__ = 'interests'

    de = Column(String(45), primary_key=True)
    en = Column(String(45), nullable=False, index=True)


@add_repr_method
class PersonalityTrait(Base):
    __tablename__ = 'personality_traits'

    de = Column(String(45), primary_key=True)
    en = Column(String(45), nullable=False, index=True)


@add_repr_method
class User(Base):
    __tablename__ = 'user'

    id = Column(INTEGER(11), primary_key=True)
    name = Column(String(45), nullable=False)
    date_of_birth = Column(DateTime, nullable=False)
    description = Column(String(45), nullable=False)
    last_online = Column(DateTime, nullable=False)
    registered = Column(DateTime, nullable=False)
    role = Column(INTEGER(11), nullable=False)
    loc_longitude = Column(DECIMAL(45, 0), nullable=False)
    loc_latitude = Column(DECIMAL(45, 0), nullable=False)
    email = Column(String(45), nullable=False)
    passwordhash = Column(String(196), nullable=False)
    gender = Column(INTEGER(11), nullable=False)
    is_email_verified = Column(TINYINT(4), nullable=False)
    modes = Column(INTEGER(11), nullable=False)


@add_repr_method
class Message(Base):
    __tablename__ = 'message'

    id = Column(INTEGER(11), primary_key=True)
    content = Column(String(45), nullable=False)
    was_sent = Column(DateTime, nullable=False)
    sender_uid = Column(ForeignKey('user.id'), nullable=False, index=True)
    receiver_uid = Column(ForeignKey('user.id'), nullable=False, index=True)
    was_read = Column(TINYINT(4), nullable=False)

    user = relationship('User', primaryjoin='Message.receiver_uid == User.id')
    user1 = relationship('User', primaryjoin='Message.sender_uid == User.id')


@add_repr_method
class UserHasInterest(Base):
    __tablename__ = 'user_has_interests'

    interests_de = Column(ForeignKey('interests.de'), primary_key=True, nullable=False, index=True)
    user_id = Column(ForeignKey('user.id'), primary_key=True, nullable=False, index=True)
    visibility = Column(INTEGER(11), nullable=False)

    interest = relationship('Interest')
    user = relationship('User')


@add_repr_method
class UserHasPersonalityTrait(Base):
    __tablename__ = 'user_has_personality_traits'

    id_user = Column(ForeignKey('user.id'), primary_key=True, nullable=False)
    id_personality_trait = Column(ForeignKey('personality_traits.de'), primary_key=True, nullable=False, index=True)
    visibility = Column(INTEGER(11), nullable=False)

    personality_trait = relationship('PersonalityTrait')
    user = relationship('User')


@add_repr_method
class Valuation(Base):
    __tablename__ = 'valuation'

    receiver_uid = Column(ForeignKey('user.id'), primary_key=True, nullable=False)
    sender_uid = Column(ForeignKey('user.id'), primary_key=True, nullable=False, index=True)
    timestamp = Column(DateTime, nullable=False)
    value = Column(INTEGER(11), nullable=False)
    remarks = Column(String(45), nullable=False)

    user = relationship('User', primaryjoin='Valuation.receiver_uid == User.id')
    user1 = relationship('User', primaryjoin='Valuation.sender_uid == User.id')

another_flask_backend\app_init_.py:

from flask import Flask
from flask_mail import Mail

from another_flask_backend.config import config
from models.interface_semantic_similarity import SemanticSimilarity
from models.wordnet_similarities import WordNetPathSim


def create_app(config_name: str) -> Flask:
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app=app)

    app.app_context().push()

    mail = Mail(app=app)
    app.config['mail'] = mail

    from another_flask_backend.app.models.db import db
    db.init_app(app)
    app.config['db'] = db

    if app.config['SSL_REDIRECT']:
        from flask_sslify import SSLify
        SSLify(app)

    from .main import main as main_blueprint
    app.register_blueprint(main_blueprint)

    from .api_v1 import api as api_blueprint
    app.register_blueprint(api_blueprint, url_prefix='/api/v1')

    similarity: SemanticSimilarity = WordNetPathSim()
    app.config['similarity'] = similarity

    return app

我错过了什么吗？我真的不知道为什么db.create_all() 不起作用。你能帮帮我吗？

编辑：我还应该提到与数据库的连接工作正常。如果我手动添加表格，我可以读写。

Answer 1

最后我在this question 找到了解决方案。由于我在数据库模型中使用declarative_base()，因此我需要使用Base.metadata.create_all(db.engine) 而不是db.create_all()。

话虽如此，完整的解决方案如下所示：

class APITestUser(unittest.TestCase):
    def setUp(self):
        self.app = create_app(Config.TESTING)
        self.app_context = self.app.app_context()
        self.app_context.push()
        Base.metadata.create_all(db.engine)
        self.client = self.app.test_client()

    def tearDown(self):
        db.session.remove()
        Base.metadata.drop_all(db.engine)
        self.app_context.pop()

Answer 2

请在db.create_all() 之后使用db.session.commit()，然后重试。 基本上，sqlalchemy 是在本地内存中创建表，但没有在数据库上执行命令。