【初心者向け】SQLAlchemyでデータベース操作をわかりやすく解説！
2024-07-27
SQLAlchemy でよくある操作をわかりやすく解説
ここでは、SQLAlchemyでよく行われる操作を、わかりやすい日本語で解説します。
データベース接続
まず、SQLAlchemyでデータベースに接続する必要があります。接続方法は、使用するデータベースの種類によって異なりますが、基本的な流れは次のとおりです。
from sqlalchemy import create_engine
# 使用するデータベースの種類と接続情報
engine = create_engine("dialect+driver://user:password@host:port/database")
このコードでは、create_engine() 関数を使用して、データベースエンジンを作成しています。引数には、使用するデータベースの種類（例：sqlite、mysql、postgresql）、接続情報（例：ユーザー名、パスワード、ホスト名、ポート番号、データベース名）を指定します。
テーブル定義
データベースに接続したら、テーブルを定義する必要があります。SQLAlchemyでは、テーブルをPythonのクラスとして定義します。
from sqlalchemy import Column, Integer, String, create_engine
# エンジンを作成
engine = create_engine("sqlite:///example.db")
# Base class for all tables
Base = declarative_base()
# ユーザー情報テーブル
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(255))
email = Column(String(255))
# テーブルの作成
Base.metadata.create_all(engine)
このコードでは、まずBaseという名前のベースクラスを作成しています。このクラスは、他のすべてのテーブルの基底クラスとなります。
次に、Userという名前のテーブルを定義しています。このテーブルには、id、name、emailという3つの列があります。
最後に、Base.metadata.create_all(engine) を実行して、定義したテーブルを実際にデータベースに作成します。
データ操作
テーブルを定義したら、データ操作を行うことができます。SQLAlchemyでは、CRUDと呼ばれる4つの操作を基本としています。
- Delete: データを削除
- Update: データを更新
- Read: データを読み出す
- Create: データを作成
Create
from sqlalchemy.orm import sessionmaker
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# 新しいユーザーを作成
user = User(name="Taro Yamada", email="[email protected]")
# セッションに追加
session.add(user)
# コミット
session.commit()
このコードでは、まずsessionmaker() 関数を使用して、セッションを作成しています。セッションは、データベースとのやり取りを行うためのオブジェクトです。
次に、User クラスのインスタンスを作成して、新しいユーザーの情報を設定しています。
最後に、session.add(user) を実行して、新しいユーザーをセッションに追加し、session.commit() を実行して、データベースに保存します。
Read
from sqlalchemy.orm import sessionmaker
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# ユーザー情報をすべて取得
users = session.query(User).all()
# 各ユーザー情報を出力
for user in users:
print(f"id: {user.id}, name: {user.name}, email: {user.email}")
このコードでは、まずsessionmaker() 関数を使用して、セッションを作成しています。
次に、session.query(User).all() を実行して、User テーブルのすべてのユーザー情報を取得しています。
最後に、取得したユーザー情報に対して、ループ処理を行い、各ユーザーの情報を個別に 출력しています。
Update
from sqlalchemy.orm import sessionmaker
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# ユーザー情報を取得
user = session.query(User).filter(User.id == 1).first()
# ユーザー名を更新
user.name = "Hanako Sato"
# コミット
session.commit()
次に、session.query(User).filter(User.id == 1).first() を実行して、id が 1 のユーザー情報を取得しています。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Text
# データベース接続
engine = create_engine("sqlite:///example.db")
Base = declarative_base()
# ユーザ情報テーブル
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(255))
email = Column(String(255))
password = Column(String(255))
# テーブルの作成
Base.metadata.create_all(engine)
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# ユーザ登録処理
def register_user(name, email, password):
# 重複登録チェック
if session.query(User).filter(User.email == email).first():
raise Exception("このメールアドレスはすでに登録されています")
# ユーザ情報の登録
user = User(name=name, email=email, password=password)
session.add(user)
session.commit()
# ユーザ登録テスト
name = "Taro Yamada"
email = "[email protected]"
password = "password123"
try:
register_user(name, email, password)
print("ユーザ登録が完了しました")
except Exception as e:
print(f"エラーが発生しました: {e}")
ログイン機能
この例では、users テーブルに登録されているユーザの認証を行うログイン機能を実装します。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Text
# データベース接続
engine = create_engine("sqlite:///example.db")
Base = declarative_base()
# ユーザ情報テーブル
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(255))
email = Column(String(255))
password = Column(String(255))
# テーブルの作成
Base.metadata.create_all(engine)
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# ログイン処理
def login(email, password):
# ユーザ情報の取得
user = session.query(User).filter(User.email == email).first()
# ユーザが存在しない場合
if not user:
return False
# パスワードが一致しない場合
if not user.verify_password(password):
return False
# ログイン成功
return True
# ログインテスト
email = "[email protected]"
password = "password123"
if login(email, password):
print("ログインに成功しました")
else:
print("ログインに失敗しました")
商品情報管理機能
この例では、products テーブルに商品情報を登録、更新、削除、検索する機能を実装します。
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Text
# データベース接続
engine = create_engine("sqlite:///example.db")
Base = declarative_base()
# 商品情報テーブル
class Product(Base):
__tablename__ = "products"
id = Column(Integer, primary_key=True)
name = Column(String(255))
price = Column(Integer)
stock = Column(Integer)
# テーブルの作成
Base.metadata.create_all(engine)
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# 商品登録処理
def register_product(name, price, stock):
product = Product(name=name, price=price, stock=stock)
session.add(product)
session.commit()
# 商品更新処理
def update_product(product_id, name, price, stock):
product = session.query(Product).filter(Product.id == product_id).first()
if product:
product.name = name
product.price = price
product.stock = stock
session.commit()
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String, Text
# データベース接続
engine = create_engine("sqlite:///example.db")
Base = declarative_base()
# ユーザ情報テーブル
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(255))
email = Column(String(255))
password = Column(String(255))
# テーブルの作成
Base.metadata.create_all(engine)
# セッションの作成
Session = sessionmaker(bind=engine)
session = Session()
# すべてのユーザ情報を取得
users = session.query(User).all()
# 各ユーザ情報を出力
for user in users:
print(f"id: {user.id}, name: {user.name}, email: {user.email}")
# 名前が "Taro" のユーザ情報を取得
taro_user = session.query(User).filter(User.name == "Taro").first()
# Taro ユーザの情報を更新
if taro_user:
taro_user.email = "taro.yamada@new_email.com"
session.commit()
コアSQL
SQLAlchemyでは、生のSQLクエリを実行することもできます。これは、パフォーマンスが重要になる場合や、SQLAlchemy の ORM では実現できない複雑な操作を実行する場合に役立ちます。
from sqlalchemy import create_engine
from sqlalchemy import text
# データベース接続
engine = create_engine("sqlite:///example.db")
# すべてのユーザ情報を取得
users = engine.execute(text("SELECT * FROM users"))
# 各ユーザ情報を出力
for user in users:
print(f"id: {user[0]}, name: {user[1]}, email: {user[2]}")
# 名前が "Taro" のユーザ情報を取得
taro_user = engine.execute(text("SELECT * FROM users WHERE name = :name"), {"name": "Taro"}).first()
# Taro ユーザの情報を更新
if taro_user:
engine.execute(text("UPDATE users SET email = :new_email WHERE id = :id"), {"new_email": "taro.yamada@new_email.com", "id": taro_user[0]})
サードパーティ製ライブラリ
sqlalchemy
