flask-sqlalchemy チートシート

flask-sqlalchemy で書き方をよく忘れるのでまとめてみた。

インストール

基本的にドキュメントを見れば ok↓
https://flask-sqlalchemy.palletsprojects.com/en/2.x/quickstart/#installation

ポイント1: 2022年6月現在、 SQLALCHEMY_TRACK_MODIFICATIONS を設定しないと deprecation warning が出る。これを入れれば ok ↓

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = True

ポイント2: postgres での database URI の書き方 (毎回忘れるので)

postgresql://postgres:YOUR_PASSWORD@localhost/YOUR_DATABASE_NAME

ちなみに最初が postgres:// ではじまってる場合エラーになるので、 .replace などを使って postgresql:// に変更する。

クエリ系

SELECT

# 全件取得
Foo.query.all()

# 1件目を取得
Foo.query.first()

# カラムで絞り込み
id = 99
all_foos = Foo.query.filter_by(id=id)

# 特定のカラムを取得
user_ids = Follower.query.with_entities(Follower.user_id).all()

# EXISTS?
exists = (
      db.session.query(Foo.id)
      .filter_by(...)
      .first() is not None
    )
if exists:
  # do something

WHERE

# 1つのカラムで絞り込み
peter = User.query.filter_by(username='peter').first()

# filter() を使う
peter = User.query.filter(User.name == 'filter').first()

# and, or を使う
from sqlalchemy import and_, or_ # flask_sqlalchemy ではない
old_people = User.query.filter(
        and_(User.birthday <= '1988-01-17', User.birthday >= '1985-01-17'))
ryan_or_british = User.query.filter(or_(db.users.name=='Ryan', db.users.country=='England'))

# datetime で絞り込みもできる
from datetime import datetime, timedelta

this_hour = datetime.now().replace(minute=0, second=0, microsecond=0)
next_hour = this_hour + timedelta(hours=1)

messages_for_this_hour = Message.query.filter(_and(created_at >= this_hour, created_at < next_hour))

filter と filter_by の違い↓
https://stackoverflow.com/a/2128558

INSERT

new_record = Foo(name="foo")
db.session.add(new_crecord)
db.session.commit()

DELETE

# 一件だけ削除
id = 99
record_to_delete = Foo.query.filter_by(id=id)
db.session.delete(record_to_delete)
db.session.commit()

# 全件削除
db.session.query(SearchCriteria).delete()
db.session.commit()

その他

# エラーが出たらロールバック
try: 
  db.session.query(SearchCriteria).delete()
  db.session.commit()
  print('deleted all')
except:
  db.session.rollback()
  print('failed to delete some records')

モデルの書き方

基本

class Student(db.Model):
  id = db.Column(db.Integer, primary_key=True) # ID はこう書く
  name = db.Column(db.String(255), nullable=False) # NOT NULL
  grade = db.Column(db.Integer)
  student_number = db.Column(db.String(255), unique=True) # UNIQUE
  created_at = db.Column(db.DateTime(timezone=True))

document: https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/#simple-example
↑よく使う型の一覧もある

テーブル名の変更

__tablename__ をモデルに追加する

class Student(db.Model):
  __tablename__ = 'students'
  id = db.Column(db.Integer, primary_key=True) # ID はこう書く
  ...

Foreign Key の書き方

db.ForeignKey('TABLE_NAME.COLUMN') で書く

class Person(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), nullable=False)
    addresses = db.relationship('Address', backref='person', lazy=True)

class Address(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(120), nullable=False)
    person_id = db.Column(db.Integer, db.ForeignKey('person.id'),
        nullable=False)

引用元: https://flask-sqlalchemy.palletsprojects.com/en/2.x/models/#one-to-many-relationships

開発系

serialize を楽にしたい

Serializer クラスを作って、モデルクラスに mixin すると便利。

class Serializer(object):
    def serialize(self):
        return {column: getattr(self, column) for column in inspect(self).attrs.keys()}

    @staticmethod
    def serialize_list(record_list):
        return [record.serialize() for record in record_list]
	
class Foo(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    ...

元ネタはここ: https://stackoverflow.com/a/27951648

SQL をログに出力してほしい

SQLALCHEMY_ECHO を設定する

from flask import Flask
app = Flask(__name__)

...

app.config['SQLALCHEMY_ECHO'] = True

document: https://flask-sqlalchemy.palletsprojects.com/en/2.x/config/

ターミナルで flask-sqlalchemy を実行したい

何もしないと RuntimeError: No application found. Either work inside a view function or push an application context. っていうエラーが出る。「アプリケーションコンテキスト」ってやつを現在のコンテキストにバインドすると使えるようになる。

# シンプルな構成
>>> from app import app 
>>> from models import db, Foo
>>> app.app_context().push() # これをやると db.session.query ... とか諸々できるようになる

# Factory パターン
>>> from factory import create_app
>>> from models import db, Foo
>>> app = create_app()
>>> app.app_context().push() 

わかりやすい日本語ブログ: https://python.ms/context/#_5-2-つのコンテキスト
document:

• https://flask.palletsprojects.com/en/2.1.x/appcontext/
• https://flask.palletsprojects.com/en/2.1.x/api/?highlight=app context push#flask.ctx.AppContext.push

ちなみに、アプリケーションコンテキストは app.py 以外から app.logger.info をするときにも使える。

# my_blueprint.py 
# Blueprint を使ってルートを複数のファイルで処理していることを想定
from flask import current_app

@my_blueprint.route('/some-endpoint')
def handle():
  current_app.logger.info('incoming request to some-endpoint')