我遇到断言错误问题。我正在尝试使用 flask_sqlalchemy 将连接集成到 gcp mySQL 实例。我已经使用另一个名为 creat_db.py 的文件中的 sqlalchemy 包成功连接到实例；但是，当我尝试将它集成到我的 flask_sqlalchemy 实现中时，我收到一个插入错误，指出 assert schema is none。这是我调用 db.create_all() 的时候。这个错误的原因是什么？
我试着去here获取文档信息，但没有找到解决方案。我也尝试过初始化我的类，例如用户和登录表单。我被困在前进的地方或检查的地方。
这是我的主文件代码。
from flask import Flask, redirect, url_for, render_template, request, flash
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timezone, timedelta
from flask_login import UserMixin, login_user, LoginManager, login_required, logout_user, current_user
from flask_wtf import FlaskForm
from werkzeug.security import generate_password_hash, check_password_hash
from wtforms import StringField, PasswordField, SubmitField, BooleanField, ValidationError
from wtforms.validators import InputRequired, Length, ValidationError, DataRequired, EqualTo, Length
from flask_bcrypt import Bcrypt
import pandas as pd
import csv
#from model_subsystem import svm
from upload_to_model import Pre_Processing_data
#import create_db
import MySQLdb
from sqlalchemy import MetaData
import create_db
import os
from dotenv import load_dotenv
load_dotenv()
DB_HOST=os.getenv('DB_HOST')
DB_PASSWORD=os.getenv('DB_PASSWORD')
DB_USER=os.getenv('DB_USER')
DB_NAME=os.getenv('DB_NAME')
DB_PORT=os.getenv('DB_PORT')
PROJECT_ID=os.getenv('PROJECT_ID')
INSTANCE_NAME=os.getenv('INSTANCE_NAME')
print(type(DB_PORT))
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = os.path.abspath(os.path.join(os.getcwd(), "service-account-key.json"))
app = Flask(__name__)
#MySQL DB
app.config["SQLALCHEMY_DATABASE_URI"]="mysql+pymysql://"
app.config['SECRET_KEY'] = 'thisisasecretkey'
app.config["SQLALCHEMY_ENGINE_OPTIONS"]={
'creator': create_db.getconn,
'pool_size': 5,
'max_overflow': 2,
'pool_timeout': 30,
'pool_recycle': 1800, }
#initialize db
db = SQLAlchemy(app)
bcrypt = Bcrypt(app)
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'signin'
@login_manager.user_loader
def load_user(user_id):
return User.query.get(int(user_id))
class User(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(50), nullable=False, unique=True)
username = db.Column(db.String(20), nullable=False, unique=True)
password_hash = db.Column(db.String(128), nullable=False)
date_added = db.Column(db.DateTime, default=datetime.utcnow)
results = db.relationship('Result', backref='poster')
def __init__(self, id, email, username, password_hash, date_added):
self.id = id
self.email = email
self.username = username
self.password_hash = password_hash
self.date_added = date_added
@property
def password(self):
raise AttributeError('Password is not a readable attribute!')
@password.setter
def password(self, password):
self.password_hash = generate_password_hash(password)
def verify_password(self, password):
return check_password_hash(self.password_hash, password)
class Result(db.Model, UserMixin):
id = db.Column(db.Integer, primary_key=True)
uid = db.Column(db.Integer, db.ForeignKey('user.id'))
type_of_cancer = db.Column(db.String(50))
model_type = db.Column(db.String(50))
model_type_of_cancer = db.Column(db.String(50))
outcome = db.Column(db.String(50))
date_created = db.Column(db.DateTime, default=datetime.utcnow)
def __init__(self, id, uid, type_of_cancer, model_type, model_type_of_cancer, outcome, date_created):
self.id = id
self.uid = uid
self.type_of_cancer = type_of_cancer
self.model_type = model_type
self.model_type_of_cancer = model_type_of_cancer
self.outcome = outcome
self.date_created = date_created
class RegisterForm(FlaskForm):
email = StringField("Email", validators=[DataRequired()], render_kw={"placeholder": "Email"})
username = StringField("Username", validators=[DataRequired()], render_kw={"placeholder": "Username"})
password_hash = PasswordField('Password', validators=[DataRequired(), EqualTo('password_hash2', message='Passwords Must Match!')], render_kw={"placeholder": "Password"})
password_hash2 = PasswordField('Confirm Password', validators=[DataRequired()], render_kw={"placeholder": "Confirm Password"})
submit = SubmitField('Register')
# def validate_username(self, username):
# existing_user_username = User.query.filter_by(
# username=username.data).first()
# if existing_user_username:
# raise ValidationError(
# 'Username already exists.')
class LoginForm(FlaskForm):
username = StringField("Username", validators=[DataRequired()], render_kw={"placeholder": "Username"})
password_hash = PasswordField("Password", validators=[DataRequired()], render_kw={"placeholder": "Password"})
submit = SubmitField("Submit")
@app.route("/")
def home():
return render_template("index.html")
@app.route("/dashboard")
@login_required
def dashboard():
id = current_user.id
results = Result.query.order_by(Result.date_created)
return render_template("dashboard.html", results=results, id=id)
@app.route("/input", methods=["GET", "POST"])
@login_required
def input():
return render_template("input.html")
@app.route('/resultpage/<int:id>')
@login_required
def resultpage(id):
userid = current_user.id
result = Result.query.get_or_404(id)
return render_template('resultpage.html', result=result, userid=userid)
@app.route("/results", methods=["GET", "POST"])
@login_required
def results():
if request.method == 'POST':
f = request.form['csvfile']
type_of_cancer = request.form['type_of_cancer']
model_type = request.form['model_type']
model_type_of_cancer = request.form['model_type_of_cancer']
results_numpy = Pre_Processing_data(f)
results_str = 'empty'
if model_type_of_cancer == 'Multiclass':
if results_numpy == [0]:
results_str='Kidney - Not Cancerous'
elif results_numpy == [1]:
results_str='Kidney - Cancerous'
elif results_numpy == [2]:
results_str='Lung - Not Cancerous'
elif results_numpy == [3]:
results_str='Lung - Cancerous'
elif results_numpy == [4]:
results_str='Breast - Not Cancerous'
elif results_numpy == [5]:
results_str='Breast - Cancerous'
else:
results_str='Model type not found'
# class Result(db.Model, UserMixin):
# id = db.Column(db.Integer, primary_key=True)
# uid = db.Column(db.Integer, db.ForeignKey('user.id'))
# type_of_cancer = db.Column(db.String(50))
# model_type = db.Column(db.String(50))
# model_type_of_cancer = db.Column(db.String(50))
# outcome = db.Column(db.String(50))
# date_created = db.Column(db.DateTime, default=datetime.utcnow)
active_user = current_user.id
#user = User(email=form.email.data, username=form.username.data, password_hash=hashed_password)
result = Result(type_of_cancer=type_of_cancer, model_type=model_type, model_type_of_cancer=model_type_of_cancer, outcome=results_str, uid=active_user)
db.session.add(result)
db.session.commit()
return render_template('results.html', model_type=model_type, type_of_cancer=type_of_cancer, results_str=results_str, model_type_of_cancer=model_type_of_cancer)
@app.route("/models")
def models():
return render_template("models.html")
@app.route("/miRNA")
def miRNA():
with open('example_data\CancerMIRNome.csv') as file:
data = []
csvfile = csv.reader(file)
for row in csvfile:
data.append(row)
data = pd.DataFrame(data)
return render_template("miRNA.html", data=data.to_html(header=False, index=False))
@app.route("/tutorial")
def tutorial():
return render_template("tutorial.html")
@app.route("/signin", methods=['GET', 'POST'])
def signin():
form = LoginForm()
if form.validate_on_submit():
user = User.query.filter_by(username=form.username.data).first()
if user:
# Check the hash
if check_password_hash(user.password_hash, form.password_hash.data):
login_user(user)
return redirect(url_for('home'))
else:
flash("Wrong Password")
else:
flash("Username not found")
return render_template("signin.html", form=form)
@app.route("/signup", methods=['GET', 'POST'])
def signup():
form = RegisterForm()
if form.validate_on_submit():
user = User.query.filter_by(email=form.email.data).first()
if user is None:
hashed_password = generate_password_hash(form.password_hash.data, "sha256")
user = User(email=form.email.data, username=form.username.data, password_hash=hashed_password)
db.session.add(user)
db.session.commit()
form.email.data = ''
form.username.data = ''
form.password_hash.data = ''
flash("Registeration successful! Please log in.")
return redirect(url_for('signin'))
return render_template("signup.html", form=form)
@app.route('/logout', methods=['GET', 'POST'])
@login_required
def logout():
logout_user()
flash('Successfully logged out.')
return redirect(url_for('signin'))
@app.route("/svmmulticlass")
def svmmulticlass():
return render_template("svmmulticlass.html")
if __name__ == "__main__":
with app.app_context():
db.create_all()
app.run(debug=True, host="0.0.0.0", port=80)
这是我的 create_db.py 文件
import os
from dotenv import load_dotenv
# Used for creating sql querys and connections
import sqlalchemy
from sqlalchemy import create_engine, Engine
# googles sql connector that uses Authorizes application with googles ADC
from google.cloud.sql.connector import Connector, IPTypes
import pymysql # is a depenency of google.cloud.sql.connector
# used to load im enviroment variables from .env file
load_dotenv()
host_cred=os.getenv('DB_HOST')
password_cred=os.getenv('DB_PASSWORD')
user_cred=os.getenv('DB_USER')
db_cred=os.getenv('DB_NAME')
INSTANCE_CONNECTION_NAME=os.getenv('INSTANCE_CONNECTION_NAME')
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = os.path.abspath(os.path.join(os.getcwd(), "service-account-key.json"))
# Creates authenticated connector
def getconn() -> pymysql.connections.Connection:
ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC
with Connector() as connector:
conn = connector.connect(
"{INSTANCE_CONNECTION_NAME}",
"pymysql",
user=user_cred,
password=password_cred,
charset='utf8',
)
return conn
# Connects to mysql via tcp socket
def connect_tcp_socket() -> sqlalchemy.engine.base.Engine:
# creates a engine object for connections
engine: Engine = create_engine(
"mysql+pymysql://",creator=getconn,#<db_user>:<db_pass>@<db_host>:<db_port>/<db_name>
pool_size=5,
max_overflow=2,
pool_timeout=30, # 30 seconds
pool_recycle=1800, # 30 minutes
)
return engine
# engine = connect_tcp_socket()
# # Shows database
# with engine.connect() as conn:
# # returns a result.cursorresult object. Hence a list of Rows
# database = conn.execute(sqlalchemy.text('CREATE DATABASE mirna'));
# print("database created")
这是我的错误和我的堆栈跟踪。
Traceback (most recent call last):
File "C:\Users\Owner.OWNER-PC\Desktop\college work\Spring2023\ED1\miRNACNet\miRNACnet.py", line 288, in <module>
db.create_all()
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\flask_sqlalchemy\extension.py", line 884, in create_all
self._call_for_binds(bind_key, "create_all")
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\flask_sqlalchemy\extension.py", line 865, in _call_for_binds
getattr(metadata, op_name)(bind=engine)
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\sql\schema.py", line 5576, in create_all
bind._run_ddl_visitor(
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\engine\base.py", line 3220, in _run_ddl_visitor
conn._run_ddl_visitor(visitorcallable, element, **kwargs)
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\engine\base.py", line 2427, in _run_ddl_visitor
visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\sql\visitors.py", line 670, in traverse_single
return meth(obj, **kw)
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\sql\ddl.py", line 903, in visit_metadata
[t for t in tables if self._can_create_table(t)]
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\sql\ddl.py", line 903, in <listcomp>
[t for t in tables if self._can_create_table(t)]
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\sql\ddl.py", line 868, in _can_create_table
return not self.checkfirst or not self.dialect.has_table(
File "<string>", line 2, in has_table
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\engine\reflection.py", line 88, in cache
return fn(self, con, *args, **kw)
File "C:\Users\Owner.OWNER-PC\AppData\Roaming\Python\Python310\site-packages\sqlalchemy\dialects\mysql\base.py", line 2696, in has_table
assert schema is not None
AssertionError
对于任何能给我指出正确方向的人谢谢！