【Flask-SQLAlchemy】INSERT/SELECT/UPDATE 操作

Flask

追加、取得、更新の操作

テーブル定義からテーブル作成

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///PFCManagement.db"
app.config["SECRET_KEY"] = os.urandom(24)
db = SQLAlchemy(app)

login_manager = LoginManager()
login_manager.init_app(app)

class USER_INFO(UserMixin, db.Model):
    id = db.Column(db.String(20), primary_key=True)
    password = db.Column(db.String(20), nullable=False)
    del_flg = db.Column(db.String(1), nullable=False, default="0")
    created_at = db.Column(
        db.DateTime, nullable=False, default=datetime.now(pytz.timezone("Asia/Tokyo"))
    )
    update_at = db.Column(db.DateTime, nullable=True)

INSERT(追加)

@app.route("/signup", methods=["GET", "POST"])
def signup():
    if request.method == "POST":
        #idをrequestから取得
        get_id = request.form.get("id")
        #passwordをrequestから取得
        get_password = request.form.get("password")
        #passwordをハッシュ化
        hashed_password = generate_password_hash(get_password, method="pbkdf2:sha256")

        #get_idとhashed_passwordでUSER_INFOのインスタンスを作成
        user_info = USER_INFO(id = get_id,
                              password = hashed_password)
        
        #作成したUSER_INFOのインスタンスをデータベースに追加し、コミット
        db.session.add(user_info)
        db.session.commit()

テーブル定義で「id」と「password」はNULL許可していないため、12行目と13行目で値の設定が必要となる

「del_flg」と「created_at」はNULL許可していないがデフォルト値を設定しているため、USER_INFOのインスタンス作成時に値の設定をしない場合、デフォルト値が設定される

「update_at」はNULL許可のため、インスタンス作成時に値の設定をしない場合、NULLが設定される

SELECT(取得)

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        #idをrequestから取得
        get_id = request.form.get("id")
        #passwordをrequestから取得
        get_password = request.form.get("password")

        #USER_INFOから「USER_INFOのid」と「requestから取得したid」が一致する最初の行を取得
        get_user_info = USER_INFO.query.filter(USER_INFO.id == get_id).first()

検索条件を設定せずに取得する方法、全件取得する方法もある
また、.all()で全件取得、.firstで最初の行を取得する

        #USER_INFOから検索条件を設定せず、全件取得
        get_user_info = USER_INFO.query.all()
        
        #USER_INFOから検索条件を設定せず、最初の行を取得
        get_user_info = USER_INFO.query.first()

        #USER_INFOから「USER_INFOのid」と「requestから取得したid」が一致する全件を取得
        get_user_info = USER_INFO.query.filter(USER_INFO.id == get_id).all()

UPDATE(更新)

@app.route("/passwordchange/<argId>", methods=["GET", "POST"])
@login_required
def passwordchange(argId):
    if request.method == "POST":
        #old_passwordをrequestから取得
        get_old_password = request.form.get("old_password")
        #new_passwordをrequestから取得
        get_new_password = request.form.get("new_password")

        try:
            #USER_INFOから「USER_INFOのid」とargIdが一致する最初の行を取得
            get_user_info = USER_INFO.query.filter(USER_INFO.id == argId).first()

            #ハッシュ化したパスワードと取得したold_passwordが一致する場合
            if check_password_hash(get_user_info.password, get_old_password):
                #取得したUSER_INFOのpasswordにrequestから取得したnew_passwordをハッシュ化し代入
                get_user_info.password = generate_password_hash(
                    get_new_password, method="pbkdf2:sha256"
                )
                #コミットする
                db.session.commit()

UPDATEしたい行を取得する
取得した行の列に値を設定する