Flask-SQLAlchemyでUPSERTを実装する方法

はじめに

サンプルアプリ(feedback-api)を用いてご紹介します。

Mac環境の記事ですが、Windows環境も同じ手順になります。環境依存の部分は読み替えてお試しください。

目的

この記事を最後まで読むと、次のことができるようになります。

No.	概要	キーワード
1	UPSERT実装	on_conflict_do_update

実行環境

環境	Ver.
macOS Catalina	10.15.3
Python	3.7.3
Flask-SQLAlchemy	2.4.1

ソースコード

実際に実装内容やソースコードを追いながら読むとより理解が深まるかと思います。是非ご活用ください。

GitHub

関連する記事

• SQLAlchemy

UPSERTの実装

UPSERTステートメントの作成

モデルにUPSERTステートメントを記述する

app/feedback/models/t_feedback.py

class Feedback(db.Model):
    __tablename__ = 't_feedback'

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    service = db.Column(db.String(255), nullable=False)
    title = db.Column(db.String(255), nullable=False)
    detail = db.Column(db.String(255), nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=created_at.default)

    """省略
    """


def upsert_stmt():
    """upsert_stmt
    """
    stmt = insert(Feedback)
    return stmt.on_conflict_do_update(
        index_elements=['id'],
        set_={
            'service': stmt.excluded.service,
            'title': stmt.excluded.title,
            'detail': stmt.excluded.detail,
            'updated_at': stmt.excluded.updated_at
        })

UPSERTステートメントのコール

コミット処理でUPSERTステートメントをコールする

app/feedback/views/api.py

"""
clause – An executable statement (i.e. an Executable expression such as select()) or string SQL statement to be executed.
params – Optional dictionary, or list of dictionaries, containing bound parameter values. If a single dictionary, single-row execution occurs; if a list of dictionaries, an “executemany” will be invoked. The keys in each dictionary must correspond to parameter names present in the statement.

e.g.) params
[
    {
        'id': 1,
        'service': 'service1',
        'title': 'title1',
        'detail': 'add as new record1'
    },
    {
        'id': 2,
        'service': 'service2',
        'title': 'title2',
        'detail': 'add as new record2'
    },
    ...
]
"""

def _upsert(payloads):
    db.session.execute(clause=upsert_stmt(), params=payloads)
    db.session.commit()