【超便利】SQLAlchemyでデータベースを操作しよう（基礎編）

from sqlalchemy import create_engine, Column, Integer, String, Float, func, select
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.exc import SQLAlchemyError

from sqlalchemy import create_engine

# SQLiteを使用した例（他のDBでもURLを調整すれば同様に使用可能）
engine = create_engine(r"sqlite:///p:/example.db", echo=True)
Base = declarative_base()
Session = sessionmaker(bind=engine)

# PostgreSQLへの接続例
user = "hoge"
password = "hoge0123"
host = "hohoho@server.com"
port = "5432"
dbname = "mydb"
driver = "postgresql+psycopg2"

engine = create_engine(f"{driver}://{user}:{password}@{host}:{port}/{dbname}")
Base = declarative_base()
Session = sessionmaker(bind=engine)

# クラスを定義
class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer)
    score = Column(Float)

# テーブル作成
Base.metadata.create_all(engine)

class User(Base):
    __tablename__ = "users"

    # カラムを定義
    id = Column(Integer, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer)
    score = Column(Float)

    # 複合主キーを定義
    __table_args__ = (
        PrimaryKeyConstraint('id', 'name'),  # 複数のカラムを主キーとして指定
    )

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False, index=True)  # name列にインデックスを作成
    age = Column(Integer)
    score = Column(Float)

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer)
    score = Column(Float)

    # 複合インデックスを定義
    __table_args__ = (
        Index('ix_name_age', 'name', 'age'),  # name と age を含む複合インデックス
    )

# インデックスの追加
Index('ix_users_name', User.name).create(bind=engine) 

with Session() as session:
    # データのインサート
    user = User(name="Alice", age=30, score=85.5)
    session.add(user)
    user = User(name="maria", age=28, score=96.7)
    session.add(user)
    session.commit()

with Session() as session:
    # データの変更
    user = session.query(User).filter_by(name="Alice").first()
    if user:
        user.age = 31
        session.commit()

with Session() as session:
    # データの削除
    user = session.query(User).filter_by(name="Alice").first()
    if user:
        session.delete(user )
        session.commit()

with Session() as session:
    try:
        # インサート
        new_user = User(name="Alice", age=30, score=85.5)
        session.add(new_user)
        new_user = User(name="maria", age=28, score=96.7)
        session.add(new_user)

        # アップデート
        user_to_update = session.query(User).filter_by(name="Alice").first()
        if user_to_update:
            user_to_update.age = 31

        # デリート
        user_to_delete = session.query(User).filter_by(name="Alice").first()
        if user_to_delete:
            session.delete(user_to_delete)

        # すべての変更を一括でコミット
        session.commit()

    except Exception as e:
        session.rollback()  # エラー時にロールバック
        print(f"エラー発生: {e}")

with Session() as session:
    results = session.query(User).filter(User.age > 25).all()
    for user in results:
        print(user.name, user.age, user.score)

# プロパティを列挙する場合
session.query(user.name, user.age, user.score)

# クラスを指定する場合
session.query(user)

# 条件式をカンマ区切りで列挙すると、AND条件として動作
filter(User.age > 25, User.score > 80)

# and_() メソッドにより明示的にAND条件で動作させることが可能
filter(and_(User.age > 25, User.score > 80))

# or_() メソッドにより明示的にOR条件で動作させることが可能
filter(or_(User.age < 18, User.score < 50))

with Session() as session:
    results = session.query(User).filter(User.age > 25)
    for user in results:
        print(user.name, user.age, user.score)

with Session() as session:
    results = session.query(User.age, func.avg(User.score)).group_by(User.age).all()
    for age, avg_score in results:
        print(f"Age: {age}, Average Score: {avg_score}")

# 新しいクラスを定義
class Address(Base):
    __tablename__ = "addresses"
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer)
    city = Column(String)

Base.metadata.create_all(engine)

# Userと Adrees を結合する
with Session() as session:
    # 結合クエリ
    results = session.query(User, Address).filter(User.id == Address.user_id).all()
    for user, address in results:
        print(user.name, address.city)

results = session.query(User, Address).join(Address, User.id == Address.user_id).all()
for user, address in results:
    print(user.name, address.city)

    results = session.query(User, Address).join(Address).all()
    for user, address in results:
        print(f"User: {user.name}, City: {address.city}")

# 並べ替え
session.query(User).order_by(User.age.desc()).all()

# 重複データの排除
session.query(User.age).distinct().all()

# 件数カウント
session.query(User).filter(User.age > 30).count()

# gropu_by と組み合わせることで、having が利用可能
session.query(User.age, func.count(User.id)).group_by(User.age).having(func.count(User.id) > 1).all()

from sqlalchemy import inspect

tables = inspect(engine).get_table_names() 
print(tables)

from sqlalchemy import inspect

inspector = inspect(engine)
columns = inspector.get_columns("users")
print(columns)

with engine.connect() as connection:
    # ストアドプロシージャの呼び出し例（MySQLなどで使用可能）
    result = connection.execute("CALL some_stored_procedure();")
    
    # SQLiteのプラグマ実行例
    connection.execute("PRAGMA foreign_keys=ON;")