原来我没有提供足够的信息来解决这个问题。问题原来是使用通过调用 SQLAlchemy 创建的 db 变量。我为名为 database.py 的数据库创建了一个 python 文件。我犯的错误是在 User\models.py 中我调用了以下 import from database import db，在 Device\models.py 中我调用了 from app import db。这导致 db.Model 无法正常运行，并且在调用 create_all() 时也不会创建用户表。希望这对将来的人有所帮助。
Database.py
from flask_influxdb import InfluxDB
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
influx_db = InfluxDB()
influx_db_client = None
def init_db():
# import all modules here that might define models so that
# they will be registered properly on the metadata. Otherwise
# you will have to import them first before calling init_db()
from users.models import User, UserStatus, UserDevice
from devices.models import Device, DeviceStatus
db.Model.metadata.drop_all(bind=db.engine)
db.Model.metadata.create_all(bind=db.engine)
Devices\models.py
from app import db
from flask_login import current_user
from sqlalchemy.orm import relationship
import enum
class DeviceStatusType(enum.Enum):
INACTIVE = "Inactive"
ACTIVE = "Active"
# Define a Device model
class Device(db.Model):
__tablename__ = 'Device'
__table_args__ = {'extend_existing': True}
id = db.Column('deviceID', db.Integer, primary_key=True)
date_created = db.Column('deviceDateCreated', db.DateTime, default=db.func.current_timestamp())
date_modified = db.Column('deviceDateModified', db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())
device_created_user = db.Column('deviceCreatedUser', db.String(128), default='App Server')
device_last_updated_user = db.Column('deviceLastUpdatedUser', db.String(128), default='App Server', onupdate=current_user)
#Serial Number
serial_number = db.Column('deviceSerialNumber', db.Integer, nullable=False, unique=True)
#Sampling Interval
sampling_interval = db.Column('deviceSamplingInterval', db.Integer, default=60, nullable=False)
# Relationship to Device Status Table
device_status_id = db.Column('deviceStatusID', db.Integer, db.ForeignKey('DeviceStatus.deviceStatusID'))
users = relationship("User", secondary="userDevice")
# New instance instantiation procedure
def __init__(self, serial_number):
self.serial_number = serial_number
self.device_status_id = 1
def __repr__(self):
return '<Device %r>' % self.serial_number
users\models.py
from random import SystemRandom
from backports.pbkdf2 import pbkdf2_hmac, compare_digest
from flask_login import UserMixin, current_user
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import relationship, backref
from devices.models import Device
import enum
# Import the database object (db) from the main application module
# We will define this inside /app/__init__.py in the next sections.
from app import db
class UserStatusType(enum.Enum):
INACTIVE = "Inactive"
ACTIVE = "Active"
# Define a User model
class User(UserMixin, db.Model):
__tablename__ = 'User'
__table_args__ = {'extend_existing': True}
id = db.Column('userID', db.Integer, primary_key=True)
date_created = db.Column('userDateCreated', db.DateTime, default=db.func.current_timestamp())
date_modified = db.Column('userDateModified', db.DateTime, default=db.func.current_timestamp(), onupdate=db.func.current_timestamp())
user_created_user = db.Column('userCreatedUser', db.String(128), default=current_user)
user_last_updated_user = db.Column('userLastUpdatedUser', db.String(128), default=current_user, onupdate=current_user)
# First Name
first_name = db.Column('userFirstName', db.String(128), nullable=False)
# Last Name
last_name = db.Column('userLastName', db.String(128), nullable=False)
# User Name
user_name = db.Column('userUserName', db.String(128), nullable=False, unique=True)
# Email
email = db.Column('userEmailAddress', db.String(128), nullable=False, unique=True)
# Password
_password = db.Column('userPassword', db.LargeBinary(128))
_salt = db.Column('userSalt', db.LargeBinary(128))
# Relationship to User Status table
user_status_id = db.Column('userStatusID', db.Integer, db.ForeignKey('UserStatus.userStatusID'))
# Relationship to UserDevice association table
devices = relationship("Device", secondary="userDevice")
@hybrid_property
def password(self):
return self._password
# In order to ensure that passwords are always stored
# hashed and salted in our database we use a descriptor
# here which will automatically hash our password
# when we provide it (i. e. user.password = "12345")
@password.setter
def password(self, value):
# When a user is first created, give them a salt
if self._salt is None:
self._salt = bytes(SystemRandom().getrandbits(8))
self._password = self._hash_password(value)
def is_valid_password(self, password):
"""Ensure that the provided password is valid.
We are using this instead of a ``sqlalchemy.types.TypeDecorator``
(which would let us write ``User.password == password`` and have the incoming
``password`` be automatically hashed in a SQLAlchemy query)
because ``compare_digest`` properly compares **all***
the characters of the hash even when they do not match in order to
avoid timing oracle side-channel attacks."""
new_hash = self._hash_password(password)
return compare_digest(new_hash, self._password)
def _hash_password(self, password):
pwd = password.encode("utf-8")
salt = bytes(self._salt)
buff = pbkdf2_hmac("sha512", pwd, salt, iterations=100000)
return bytes(buff)
# New instance instantiation procedure
def __init__(self, first_name, last_name, user_name, email, password):
self.first_name = first_name
self.last_name = last_name
self.user_name = user_name
self.email = email
self.password = password
self.user_status_id = 2
def __repr__(self):
return "<User #{:d}>".format(self.id)