Go to Qiita Advent Calendar 2024 Top
LoginSignup
25
25

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 3 years have passed since last update.

@penpenta

SQLAlchemyを簡単にまとめてみた（TIPSもあるよ）

Last updated at Posted at 2020-05-06

目標

python の SQL ツールキットである SQLAlchemy のまとめ
質問や指摘・編集リクエストは大歓迎です。よろしくお願いします。

SQLAlchemy とは？？

Python で DB を操作するときに使用する ORM（Object Relational Mapper）ツール

環境準備

Docker で環境を構築

手軽に Docker で構築する

docker-compose.yml
version: "3"

services:
  db:
    image: postgres:11.7
    container_name: sqlalchemy-db
    ports:
      - 5432:5432
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
      - POSTGRES_DB=sqlalchemy
  app:
    image: python:3.8.2-slim
    container_name: sqlalchemy-app
    volumes:
      - ./src:/usr/src
    environment:
      - PYTHONPATH=${PYTHONPATH}:/usr/src
    tty: true

パッケージのインストール

SQLAlchemy と psycopg2-binary のインストールを行う

pythonのパッケージのインストール
pip install SQLAlchemy psycopg2-binary

root@4e7436ca992b:/usr/src# pip install SQLAlchemy psycopg2-binary
Collecting SQLAlchemy
  Downloading SQLAlchemy-1.3.16-cp38-cp38-manylinux2010_x86_64.whl (1.2 MB)
     |████████████████████████████████| 1.2 MB 8.3 MB/s
Collecting psycopg2-binary
  Downloading psycopg2_binary-2.8.5-cp38-cp38-manylinux1_x86_64.whl (3.0 MB)
     |████████████████████████████████| 3.0 MB 60.6 MB/s
Installing collected packages: SQLAlchemy, psycopg2-binary
Successfully installed SQLAlchemy-1.3.16 psycopg2-binary-2.8.5

SQLAlchemy の準備

ここから SQLAlchemy の準備を行う
簡単にイメージを補足すると、以下の様なイメージになる

Engine は Python と DB（PostgresSQL）を繋ぐ橋
Session は Engine を利用して DB に実行してもらう 命令（Query） を詰め込む箱みたいなもの
Models は DB で定義してあるテーブルの情報と紐づけるためのクラス
また、各 Model はdeclarative_baseで定義したクラスを継承する

Engine と Session の作成

Engine の作成には create_engine() を使用する
Session の作成には sessionmaker()を使用する

setting.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

# Engine の作成
Engine = create_engine(
    "postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy",
    encoding="utf-8",
    echo=False
)

# Session の作成
session = sessionmaker(
    autocommit=False,
    autoflush=False,
    bind=Engine
)

BaseModel = declarative_base()

Model の作成

UserModel を定義する
ここでは id と name の列を定義するのみとする

model.py
from sqlalchemy import Column, String, Integer
from setting import BaseModel


class UserModel(BaseModel):
    """
    UserModel
    """
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(255), nullable=False)

    def __init__(self, name, created_at=None, updated_at=None):
        self.name = name

SQLAlchemy の利用

BaseModel を使用してテーブルの作成と削除

BaseModel を全てのモデルで継承していると BaseModel の metadata を利用してテーブルの作成と削除ができる
単体テストを書くときなどに SetUp と TearDown によく書くもの

テーブルの作成と削除
from setting import Engine
from model import BaseModel

# テーブルの作成
BaseModel.metadata.create_all(bind=Engine)

# テーブルの削除
BaseModel.metadata.drop_all(Engine)

レコードの挿入（INSERT 文）

INSERT 文は session.add() を使用する

  • 変更の完了：session.commit()
  • 変更の取り消し：session.rollback()
  • session の使用終了：session.close()

SQLAlchemyError は SQLAlchemy 内で発生したエラーを全てキャッチしてくれるもの

レコードの挿入（INSERT文）
from setting import Session, Engine
from model import UserModel, BaseModel
from sqlalchemy.exc import SQLAlchemyError

def model_insert():

    try:
        session = Session()

        user = UserModel(name='マスオ')

        session.add(user)

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

if __name__ == "__main__":
    # テーブルの作成
    BaseModel.metadata.create_all(bind=Engine)

    model_insert()

    # テーブルの削除
    BaseModel.metadata.drop_all(Engine)

レコードの取得（SELECT 文）

session の query()　を使用してレコードを取得する

  • １レコードのみ取得：first() を使用
  • 全てのレコードを取得：all() を使用
  • 条件指定：filter() を使用

他にも色々（テーブルの結合など）あるので下記のチュートリアルから探していくと良いかも
Object Relational Tutorial — SQLAlchemy 1.3 Documentation

レコードの取得（SELECT文）
def model_select():

    try:
        session = Session()

        # １レコードのみ取得
        user = session.query(UserModel).first()
        # 全てのレコードを取得
        user = session.query(UserModel).all()
        # 条件を指定して取得（WHERE句）
        user = session.query(UserModel).\
            filter(UserModel.name == 'マスオ').\
            all()

    except SQLAlchemyError:
        pass
    finally:
        session.close()

レコードの更新（UPDATE 文）

session で取得したレコードのプロパティを書き換えて commit すると更新される

レコードの更新（UPDATE文）
def model_update():

    try:
        session = Session()

        # 条件を指定して取得（WHERE句）
        user = session.query(UserModel).\
            filter(UserModel.name == 'マスオ').\
            first()

        user.name = 'タラオ'

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

レコードの削除（DELETE 文）

session で取得したレコードに対して delete() を実行すると削除される

レコードの更新（UPDATE文）
def model_delete():

    try:
        session = Session()

        session.query(UserModel).\
            filter(UserModel.name == 'タラオ').\
            .delete()

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()

ソースコードまとめ

main.py
from setting import Session, Engine
from model import UserModel, BaseModel
from sqlalchemy.exc import SQLAlchemyError


def model_insert():

    try:
        session = Session()

        user = UserModel(name='マスオ')

        session.add(user)

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


def model_select():

    try:
        session = Session()

        # １レコードのみ取得
        user = session.query(UserModel).first()
        # 全てのレコードを取得
        user = session.query(UserModel).all()
        # 条件を指定して取得（WHERE句）
        user = session.query(UserModel).\
            filter(UserModel.name == 'マスオ').\
            all()

    except SQLAlchemyError:
        pass
    finally:
        session.close()


def model_update():

    try:
        session = Session()

        # 条件を指定して取得（WHERE句）
        user = session.query(UserModel).\
            filter(UserModel.name == 'マスオ').\
            first()

        user.name = 'タラオ'

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


def model_delete():

    try:
        session = Session()

        # 条件を指定して取得（WHERE句）
        session.query(UserModel).\
            filter(UserModel.name == 'タラオ').\
            .delete()

        session.commit()

    except SQLAlchemyError:
        session.rollback()
    finally:
        session.close()


if __name__ == "__main__":
    # テーブルの作成
    BaseModel.metadata.create_all(bind=Engine)

    model_insert()

    model_select()

    model_update()

    model_delete()

    # テーブルの削除
    BaseModel.metadata.drop_all(Engine)

Tips

共通の列を持たせたいのですが？？

例えば生成日付（created_at）と更新日付（updated_at）とか
毎回、モデルの定義に書くのは大変なので @declared_attr を使用すると便利
@declared_attr を使用して作成した列を持つクラスを BaseModel の作成時に引き渡すと共通の列を作成するようになる

共通の列を作成
class Base(object):
    @declared_attr
    def created_at(cls):
        return Column(DateTime, default=datetime.now, nullable=False)

    @declared_attr
    def updated_at(cls):
        return Column(DateTime, default=datetime.now, nullable=False)

BaseModel = declarative_base(cls=Base)

DB の作成はできないの？？

SQLAlchemy-Utils — SQLAlchemy-Utils 0.33.9 documentationを使用すると簡単に作成や削除ができる
テストの最初やアプリケーションの起動時に使用するイメージ

DBの存在チェックと作成
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database
engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
    if not database_exists(engine.url):
        # DBの作成
        create_database(engine.url)
    # DBの削除
    drop_database(engine.url)

スキーマの作成は？？

postgresなどを使用していてスキーマを作成したい場合は、CreateSchema と DropSchema を使用する

スキーマの作成・削除
from sqlalchemy.schema import CreateSchema, DropSchema
from sqlalchemy import create_engine

engine = create_engine("postgresql://postgres:postgres@sqlalchemy-db:5432/sqlalchemy")
schema_name = 'test'
# スキーマの存在チェック
if not engine.dialect.has_schema(Engine, schema_name):
    # スキーマの作成
    engine.execute(CreateSchema(schema_name))

# スキーマの削除
Engine.execute(DropSchema(schema_name))
25
25
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
25
25

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?