SQLAlchemy2.0でCRUD操作を行う

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base 

# エンジンのセットアップ
DATABASE_URL = "your_database_url"
engine = create_engine(DATABASE_URL, echo=True)

# ベースクラスを作成
Base = declarative_base()

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship


# Userテーブルの定義
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, autoincrement=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    items = relationship("Item", back_populates="owner")

# Itemテーブルの定義
class Item(Base):
    __tablename__ = "items"
    
    id = Column(Integer, primary_key=True, autoincrement=True, index=True)
    title = Column(String, index=True)
    description = Column(String)
    owner_id = Column(Integer, ForeignKey("users.id"))
    
    owner = relationship("User", back_populates="items")

from contextlib import contextmanager


# セッションの作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# get_db関数の定義
# contextmanagerはリソースの確保と解放の安全を保つため。
# フレームワーク（DjangoやFlaskなど）では不要
@contextmanager
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from sqlalchemy.orm import Session
from sqlalchemy.exc import SQLAlchemyError

# 共通のヘルパー関数。主にinsert, update, delete文で使う
def execute_handler(execution, db: Session):
  try:
    result = execution()
    if not result:
      return None
    db.commit()
    return result
  except SQLAlchemyError as e:
    print(e)
    db.rollback()  # エラー発生時にはロールバック
    return None
  except Exception as e:
    print(e)
    db.rollback()  # エラー発生時にはロールバック
    return None

from sqlalchemy import insert, select, update, delete, join
from sqlalchemy.orm import Session


def create_user(new_user: dict, db: Session):
  stmt = insert(User).values(new_user)
  create_meta = execute_handler(lambda: db.execute(stmt), db)
  return create_meta

def get_user(id: int, db: Session):
  stmt = select(User).where(User.id == id)
  user = db.execute(stmt).first()
  return user

def get_all_users(db: Session):
  stmt = select(User)
  user_list = db.execute(stmt).all()
  return user_list

def get_users_with_items(db: Session):
  # UserテーブルとItemテーブルをjoinして取得する
  stmt = select(User, Item
      ).select_from(
        join(
            User, 
            Item,
            User.id == Item.owner_id
        )
      )
  user_with_item_list = db.execute(stmt).all()
  return user_with_item_list
  
def update_user(id: int, values: dict, db: Session) -> bool:
  stmt = update(User).where(User.id == id).values(values)
  update_meta = execute_handler(lambda: db.execute(stmt), db)
  if update_meta is not None:
    return True
  return False
  
def delete_user(id: int, db: Session) -> bool:
  stmt = delete(delete).where(User.id == id)
  delete_meta = execute_handler(lambda: db.execute(stmt), db)
  if delete_meta is not None:
    return True
  return False

from sqlalchemy import insert, select, update, delete
from sqlalchemy.orm import Session

def create_item(new_item: dict, db: Session):
  stmt = insert(Item).values(new_item)
  create_meta = execute_handler(lambda: db.execute(stmt), db)
  return create_meta

def get_item(id: int, db: Session):
  stmt = select(Item).where(Item.id == id)
  item = db.execute(stmt).first()
  return item

def get_all_items(db: Session):
  stmt = select(Item)
  item_list = db.execute(stmt).all()
  return item_list
  
def update_item(id: int, values: dict, db: Session) -> bool:
  stmt = update(Item).where(Item.id == id).values(values)
  update_meta = execute_handler(lambda: db.execute(stmt), db)
  if update_meta is not None:
    return True
  return False
  
def delete_item(id: int, db: Session) -> bool:
  stmt = delete(delete).where(Item.id == id)
  delete_meta = execute_handler(lambda: db.execute(stmt), db)
  if delete_meta is not None:
    return True
  return False