python SQLAlchemy データベース追加、表示

Python

前回の記事でSQLAlchemyでSQLiteのデータベースを作成するコードを紹介しました

python SQLAlchemy データベース連携
python SQLAlchemyを使ってSQLiteのデータベースを操作するサンプルコードを解説します
go-pro-world.net
2025.01.11

今回は作成したデータベースに追加で編集するコードを紹介していきます

・データベースに追加

既にあるデータベースのテーブルに、辞書型データを追加するコードです

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# データベース接続設定
DATABASE_URL = "sqlite:///example.db"  # SQLiteを使用
engine = create_engine(DATABASE_URL, echo=True)

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

# テーブル定義
class ExampleTable(Base):
    __tablename__ = 'example_table'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)
    score = Column(Float, nullable=False)

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

# データを追加するコード
Session = sessionmaker(bind=engine)
session = Session()

# 追加するデータのリスト
data_to_add = [
    {"name": "Alice", "age": 25, "score": 85.5},
    {"name": "Bob", "age": 30, "score": 90.0},
    {"name": "Charlie", "age": 22, "score": 78.0}
]

try:
    # データをテーブルに追加
    for data in data_to_add:
        record = ExampleTable(
            name=data['name'],
            age=data['age'],
            score=data['score']
        )
        session.add(record)  # セッションに追加
    session.commit()  # トランザクションを確定
    print("データが正常に追加されました！")
except Exception as e:
    session.rollback()  # エラー発生時にロールバック
    print(f"エラーが発生しました: {e}")
finally:
    session.close()  # セッションを閉じる

途中までは同じですが、今回はデータフレームでなく辞書型データをforで回してrecordに追加しています
前回の内容が分かれば難しくないと思います

・データベースを表示

既にあるデータベースのテーブルのレコードを表示するコードです

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# データベース接続設定
DATABASE_URL = "sqlite:///example.db"  # SQLiteを使用
engine = create_engine(DATABASE_URL, echo=True)

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

# テーブル定義
class ExampleTable(Base):
    __tablename__ = 'example_table'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)
    score = Column(Float, nullable=False)

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

# データベース接続設定（既存のエンジンを使用）
Session = sessionmaker(bind=engine)
session = Session()

# データベースを表示する部分
try:
    # テーブルの全データを取得
    records = session.query(ExampleTable).all()

    # データを表示
    print("現在のデータベースの内容:")
    for record in records:
        print(f"ID: {record.id}, Name: {record.name}, Age: {record.age}, Score: {record.score}")

except Exception as e:
    print(f"エラーが発生しました: {e}")

finally:
    # セッションを閉じる
    session.close()

初めて出てきたコードの部分を解説します

records = session.query(ExampleTable).all()

  • session.query(ExampleTable):

ExampleTableクラスに関連付けられたテーブルのデータをクエリします

  • .all():

クエリ結果として返されたすべてのレコードをリストとして取得します

ここで.filterを使って検索結果を絞ることもできます

records = session.query(ExampleTable).filter(ExampleTable.age >= 25).all()

  • .filter(ExampleTable.age >= 25)
    ageの値が25より大きいものだけ、recordsに格納します

for record in records:
print(f”ID: {record.id}, Name: {record.name}, Age: {record.age}, Score: {record.score}”)

レコードをリスト形式で取得できたので、ループで処理しそれぞれのフィールドを出力します

・データベースを削除

データベースからレコードを削除するコードです

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# データベース接続設定
DATABASE_URL = "sqlite:///example.db"  # SQLiteを使用
engine = create_engine(DATABASE_URL, echo=True)

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

# テーブル定義
class ExampleTable(Base):
    __tablename__ = 'example_table'
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)
    score = Column(Float, nullable=False)

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

# セッション作成
Session = sessionmaker(bind=engine)
session = Session()

try:
    # 条件に一致するレコードを取得
    record_to_delete = session.query(ExampleTable).filter(ExampleTable.name == "Alice").first()
   
    # レコードが存在する場合に削除
    if record_to_delete:
        session.delete(record_to_delete)
        session.commit()
        print("レコードを削除しました！")
    else:
        print("削除対象のレコードが見つかりませんでした。")

except Exception as e:
    session.rollback()
    print(f"エラーが発生しました: {e}")
finally:
    session.close()

record_to_delete = session.query(ExampleTable).filter(ExampleTable.name == “Alice”).first()

レコードの指定方法はデータベースを表示する方法とほぼ同じです
今回は.first()をつかって最初にヒットしたレコードのみを指定しています

session.delete(record_to_delete)

.deleteメソッドでレコードを削除します

コメント