【Python】SqlalchemyでのテーブルからのSELECT処理速度を高速化する

今回は前回に引き続きSQLAlchemyを使ってPythonでデータベースの情報をSELECTで抽出する処理を高速化すう知見についてまとめていきます。

前回の記事：【Python】SqlalchemyでのINSERT処理を高速化する方法

Contents [hide]

前準備
SqlalchemyでのSELECT文の実行
- 通常の方法
- SQLAlchemy Coreを使う方法

終わり

前準備

前回と同じ要領でテーブルを作成し、1000レコードのデータを登録します。

from datetime import datetime

from sqlalchemy import create_engine, MetaData

from sqlalchemy.orm import scoped_session, sessionmaker

import time

uri = 'sqlite:///test.db'

engine = create_engine(uri)

from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

from sqlalchemy.schema import Column

from sqlalchemy.types import Integer, String

class User(Base):
    __tablename__ = "user"  # テーブル名を指定
    user_id = Column(Integer, primary_key=True)
    first_name = Column(String(255))
    last_name = Column(String(255))
    age = Column(Integer)

    def full_name(self):  # フルネームを返すメソッド
        return "{self.first_name} {self.last_name}"


Base.metadata.create_all(engine)

from sqlalchemy.orm import sessionmaker

SessionClass = sessionmaker(engine)

session = SessionClass()

user_a = User(first_name="first_a", last_name="last_a", age=20)

session.add(user_a)

session.commit()

# 登録用のデータを作成

data_list = []

for i in range(1000):
    data_list.append(["first_a","last_a", i])

#bulk_insert

session.bulk_save_objects(
    [User(first_name=d[0],
               last_name=d[1],
               age=d[2])
     for d in data_list], return_defaults=True)

session.commit()

SqlalchemyでのSELECT文の実行

通常の方法

from sqlalchemy import select, desc, and_, func 

start_core = time.time() 

users = session.query(User).order_by(desc(User.age)).limit(100).all() 

result = [{'first_name': u.first_name, 'last_name': u.last_name, 'age': u.age} for u in users] 

print(time.time() -start_core)

実行時間：0.028秒

SQLAlchemy Coreを使う方法

次はSQLAlchemy Coreを使ってSELECT文を実行してみます。

コードは以下のようになります。

start_core = time.time() 

u = User.__table__.c sel = select([u.first_name, u.last_name, u.age])\ 

.select_from(User.__table__)\ 

.order_by(desc(u.age)).limit(100)

result = [{'first_name': r[0], 'last_name': r[1], 'age': r[2]} for r in session.execute(sel)]

print(time.time() -start_core)