Flaskアプリにて、MySQLに接続し、flask db migrateコマンドを実行したいです。
解決したいこと
SQLAlchemyを利用してFlaskからRDSのMySQLに接続し、flask db migrateコマンドを実行したいです。
前提
Python Flask による Web アプリ開発入門という本で出てくるアプリを基に作成しています。
・FlaskでWebアプリをつくっています。
・Flask-SQLAlchemyを利用しています。
・flask db migrateを実行した際に、エラーになってしまいます。
・以前、SQLiteをDBとして利用してた際には、flask db migrateコマンドが実行できました。
・RDSのMySQLに変更後できなくなりました。
・MySQL Workbechやvscodeの拡張機能からは接続できました。
発生している問題・エラー
sqlalchemy.exc.OperationalError: (pymysql.err.OperationalError) (1045, "Access denied for user 'ユーザー名'@'自宅のIPアドレスのようなもの' (using password: YES)")
(Background on this error at: https://sqlalche.me/e/20/e3q8)
該当するソースコード
app.py
from flask import Flask
from flask_login import LoginManager
from flask_migrate import Migrate
from flask_sqlalchemy import SQLAlchemy
from flask_wtf.csrf import CSRFProtect
from apps.config import config
db = SQLAlchemy()
csrf = CSRFProtect()
login_manager = LoginManager()
login_manager.login_view = "auth.signup"
login_manager.login_message = ""
def create_app(config_key):
app = Flask(__name__)
app.config.from_object(config[config_key])
csrf.init_app(app)
db.init_app(app)
Migrate(app, db)
login_manager.init_app(app)
from apps.crud import views as crud_views
from apps.auth import views as auth_views
from apps.scrapy import views as sc_views
app.register_blueprint(crud_views.crud, url_prefix="/crud")
app.register_blueprint(auth_views.auth, url_prefix="/auth")
app.register_blueprint(sc_views.sc)
return app
config.py
from pathlib import Path
import secrets
basedir = Path(__file__).parent.parent
class BaseConfig:
SECRET_KEY = secrets.token_urlsafe(32)
WTF_CSRF_SECRET_KEY = secrets.token_urlsafe(40)
class LocalConfig(BaseConfig):
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://ユーザー名:パスワード@エンドポイント:3306/DB名'
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO=True
class TestingConfig(BaseConfig):
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://ユーザー名:パスワード@エンドポイント:3306/DB名'
SQLALCHEMY_TRACK_MODIFICATIONS = False
WTF_CSRF_ENABLED = False
config = {
"testing": TestingConfig,
"local": LocalConfig
}
models.py
from datetime import datetime
from apps.app import db, login_manager
from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
class User(UserMixin, db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True, nullable=False)
username = db.Column(db.String(80), index=True)
email = db.Column(db.String(120), unique=True, index=True)
password_hash = db.Column(db.String(128))
created_at = db.Column(db.DateTime, default=datetime.now)
updated_at = db.Column(db.DateTime, default=datetime.now, onupdate=datetime.now)
@property
def password(self):
raise AttributeError("読み取り不可")
@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)
def is_duplicate_email(self):
return User.query.filter_by(email=self.email).first() is not None
@login_manager.user_loader
def load_user(user_id):
return User.query.get(user_id)
自分で試したこと
pythonからMySQLへの接続に利用するDBAPIは一通りインストールしました。
MySQL WorkbenchでMySQL上にDBを作製し、問題なくflaskアプリを起動できました。
下記コードを実行することでMySQL上にDBを作製することができました。
#create_db.py
import mysql.connector
mydb = mysql.connector.connect(
host = RDSのエンドポイント,
user = ユーザー名,
password = パスワード,
)
my_cursor = mydb.cursor()
my_cursor.execute("CREATE DATABASE flask_db")
my_cursor.execute("SHOW DATABASES")
for db in my_cursor:
print(db)
0
