LoginSignup
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@k2-gc

SQLite3 と SQLAlchemy で同じことをする

Posted at

SQLite3 と SQLAlchemy で同じことをする

はじめに

これまで SQLite3 で DB を操作していましたが，やはり SQLAlchemy が触りやすいなと思い，今更ながら，基礎的な DB 操作についての比較をしてみました．

SQLAlchemy とは

  1. Python 用の SQL ツール
  2. Object Relational Mapper（ORM）を可能にする

    👉 生の SQL を書かずに、Python のクラスやメソッドで DB を操作できるようにする

ORM とは

簡単にいうと，クラスと DB のテーブルを紐づける技術

ここから，SQL の基礎的な操作を SQLAlchemy と SQLite3 で書いてみます．

Step1：DB 作成とお試し操作

まずは単純な DB 作成と適当なデータの挿入，取得です．

SQLAlchemy

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

Base = declarative_base()
db_path = "mydb.db"

"""
Model Definition
"""

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)


"""
Engine and Table Creation
"""

engine = create_engine(f'sqlite:///{db_path}', echo=True)
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

"""
Session Creation and Data add
"""

Session = sessionmaker(bind=engine)
session = Session()

user = User(name="Alice", email="alice@example.com")
session.add(user)
session.commit()

result = session.query(User).filter_by(name="Alice").first()
print(result.id, result.name, result.email)

SQLite3

import sqlite3

# DB接続とカーソル作成
conn = sqlite3.connect('mydb_raw.db')
cur = conn.cursor()

# テーブル作成（なければ）
cur.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    name TEXT,
    email TEXT
)
''')

# データ追加
cur.execute('''
INSERT INTO users (name, email) VALUES (?, ?)
''', ("Alice", "alice@example.com"))

conn.commit()

# データ取得
cur.execute('''
SELECT * FROM users WHERE name = ?
''', ("Alice",))
user = cur.fetchone()
print(user)

conn.close()

Step2：CRUD 操作

CREATE/READ/UPDATE/DELETE を行います．

SQLAlchemy

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

Base = declarative_base()
db_path = "mydb.db"

"""
Model Definition
"""

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)


"""
Engine and Table Creation
"""

engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

"""
Session Creation and Data add
"""

Session = sessionmaker(bind=engine)
session = Session()

user = User(name="Alice", email="alice@example.com")
session.add(user)
session.commit()


"""
CRUD
"""

"""
CREATE/INSERT
"""

print("*"*10, "CREATE", "*"*10)
# Single
new_user = User(name="Bob", email="bob@example.com")
session.add(new_user)
session.commit()

# Multi
session.add_all([
    User(name="Charlie", email="charlie@example.com"),
    User(name="Dana", email="dana@example.com"),
])
session.commit()

"""
READ/SELECT
"""
print("*"*10, "READ", "*"*10)
users = session.query(User).all()
for user in users:
    print(user.id, user.name, user.email)

user = session.query(User).filter_by(name="Alice").first()
print(user.id, user.name, user.email)
user = session.query(User).filter(User.name == "Alice").first()
print(user.id, user.name, user.email)


"""
UPDATE
"""

print("*"*10, "UPDATE", "*"*10)
user = session.query(User).filter_by(name="Bob").first()
user.email = "newbob@example.com"
session.commit()

users = session.query(User).all()
for user in users:
    print(user.id, user.name, user.email)

"""
DELETE
"""
print("*"*10, "DELETE", "*"*10)
user = session.query(User).filter_by(name="Charlie").first()
session.delete(user)
session.commit()
users = session.query(User).all()
for user in users:
    print(user.id, user.name, user.email)

SQLite3

import sqlite3

# DB接続 & カーソル作成
conn = sqlite3.connect("mydb_raw.db")
cur = conn.cursor()

# テーブル作成（なければ）
cur.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    email TEXT
)
''')

"""
CREATE/INSERT
"""
print("*" * 10, "CREATE", "*" * 10)
# Single
cur.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("Bob", "bob@example.com"))

# Multi
cur.executemany("INSERT INTO users (name, email) VALUES (?, ?)", [
    ("Charlie", "charlie@example.com"),
    ("Dana", "dana@example.com"),
])

conn.commit()

"""
READ/SELECT
"""
print("*" * 10, "READ", "*" * 10)
cur.execute("SELECT * FROM users")
for row in cur.fetchall():
    print(row)

cur.execute("SELECT * FROM users WHERE name = ?", ("Alice",))
user = cur.fetchone()
if user:
    print(user)

cur.execute("SELECT * FROM users WHERE name = ?", ("Alice",))
user = cur.fetchone()
if user:
    print(user)

"""
UPDATE
"""
print("*" * 10, "UPDATE", "*" * 10)
cur.execute("UPDATE users SET email = ? WHERE name = ?", ("newbob@example.com", "Bob"))
conn.commit()

cur.execute("SELECT * FROM users")
for row in cur.fetchall():
    print(row)

"""
DELETE
"""
print("*" * 10, "DELETE", "*" * 10)
cur.execute("DELETE FROM users WHERE name = ?", ("Charlie",))
conn.commit()

cur.execute("SELECT * FROM users")
for row in cur.fetchall():
    print(row)

# 後処理
conn.close()

Step3-1：リレーション（One-to-Many）

リレーションの1対多です．

SQLAlchemy

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

Base = declarative_base()
db_path = "mydb.db"

"""
Model Definition
"""

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))

    user = relationship("User", back_populates="posts")

"""
Engine and Table Creation
"""

engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

"""
Session Creation
"""

Session = sessionmaker(bind=engine)
session = Session()

# User Creation
user = User(name="Alice")
session.add(user)
session.commit()

# Add Post
post1 = Post(title="Hello, World", user_id=user.id)
post2 = Post(title="Another post", user_id=user.id)

session.add_all([post1, post2])
session.commit()

# Another Way
user.posts.append(Post(title="Third post"))
session.commit()

# Check from User
print("*" * 10, "USER -> POSTS", "*" * 10)
user = session.query(User).filter_by(name="Alice").first()
for post in user.posts:
    print(post.title)

# Check from post
print("*" * 10, "POST -> USER", "*" * 10)
post = session.query(Post).filter_by(title="Hello, World").first()
print(post.title, "by", post.user.name)

SQLite3

import sqlite3

conn = sqlite3.connect("mydb_raw.db")
cur = conn.cursor()

# テーブル作成
cur.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT
)
''')

cur.execute('''
CREATE TABLE IF NOT EXISTS posts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT,
    user_id INTEGER,
    FOREIGN KEY(user_id) REFERENCES users(id)
)
''')

# ユーザー作成
cur.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
user_id = cur.lastrowid  # ← 自動採番されたIDを取得

# 投稿追加（方法1：user_id指定）
cur.executemany("INSERT INTO posts (title, user_id) VALUES (?, ?)", [
    ("Hello, World", user_id),
    ("Another post", user_id)
])

# 投稿追加（方法2：あとからuser_id指定して追加）
cur.execute("INSERT INTO posts (title, user_id) VALUES (?, ?)", ("Third post", user_id))

conn.commit()

# USER → POSTS
print("*" * 10, "USER -> POSTS", "*" * 10)
cur.execute("SELECT title FROM posts WHERE user_id = ?", (user_id,))
for row in cur.fetchall():
    print(row[0])

# POST → USER
print("*" * 10, "POST -> USER", "*" * 10)
cur.execute("SELECT users.name, posts.title FROM posts JOIN users ON posts.user_id = users.id WHERE posts.title = ?", ("Hello, World",))
row = cur.fetchone()
print(f"{row[1]} by {row[0]}")

conn.close()

Step3-2：リレーション（Many-to-Many）

リレーションの多対多です．

SQLAlchemy

from sqlalchemy import Column, Integer, String, Table, create_engine, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()
db_path = "mydb.db"


"""
Association
"""

user_group_table = Table(
    'user_group',
    Base.metadata,
    Column('user_id', Integer, ForeignKey('users.id')),
    Column('group_id', Integer, ForeignKey('groups.id'))
)

"""
Model Definition
"""

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

    groups = relationship("Group", secondary=user_group_table, back_populates="users")

class Group(Base):
    __tablename__ = 'groups'
    id = Column(Integer, primary_key=True)
    name = Column(String)

    users = relationship("User", secondary=user_group_table, back_populates="groups")

"""
Engine and Table Creation
"""

engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

"""
Session Creation and Data add
"""

Session = sessionmaker(bind=engine)
session = Session()

user1 = User(name="Alice", email="alice@example.com")
user2 = User(name="Bob", email="bob@example.com")
group1 = Group(name="Admin")
group2 = Group(name="Users")

# M to M
user1.groups.append(group1) # Alice <-> Admin
user1.groups.append(group2) # Alice <-> Users
user2.groups.append(group2) # Bob <-> Users

session.add_all([user1, user2, group1, group2])
session.commit()

# ユーザーからグループを確認
print("*" * 10, "USER → GROUP", "*" * 10)
for group in user1.groups:
    print(f"{user1.name} belongs to group: {group.name}")

# グループからユーザーを確認
print("*" * 10, "GROUP → USER", "*" * 10)
for user in group2.users:
    print(f"Group {group2.name} has member: {user.name}")

SQLite3

import sqlite3

conn = sqlite3.connect("mydb_raw.db")
cur = conn.cursor()

# ユーザーテーブル
cur.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT,
    email TEXT
)
''')

# グループテーブル
cur.execute('''
CREATE TABLE IF NOT EXISTS groups (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT
)
''')

# 中間テーブル（多対多のつなぎ役）
cur.execute('''
CREATE TABLE IF NOT EXISTS user_group (
    user_id INTEGER,
    group_id INTEGER,
    FOREIGN KEY(user_id) REFERENCES users(id),
    FOREIGN KEY(group_id) REFERENCES groups(id)
)
''')

# ユーザー追加
cur.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("Alice", "alice@example.com"))
user1_id = cur.lastrowid
cur.execute("INSERT INTO users (name, email) VALUES (?, ?)", ("Bob", "bob@example.com"))
user2_id = cur.lastrowid

# グループ追加
cur.execute("INSERT INTO groups (name) VALUES (?)", ("Admin",))
group1_id = cur.lastrowid
cur.execute("INSERT INTO groups (name) VALUES (?)", ("Users",))
group2_id = cur.lastrowid

# 多対多リレーション（user_group に登録）
cur.executemany("INSERT INTO user_group (user_id, group_id) VALUES (?, ?)", [
    (user1_id, group1_id),  # Alice - Admin
    (user1_id, group2_id),  # Alice - Users
    (user2_id, group2_id)   # Bob - Users
])

conn.commit()

# USER → GROUP
print("*" * 10, "USER → GROUP", "*" * 10)
cur.execute('''
SELECT g.name
FROM groups g
JOIN user_group ug ON g.id = ug.group_id
WHERE ug.user_id = ?
''', (user1_id,))
for row in cur.fetchall():
    print(f"Alice belongs to group: {row[0]}")

# GROUP → USER
print("*" * 10, "GROUP → USER", "*" * 10)
cur.execute('''
SELECT u.name
FROM users u
JOIN user_group ug ON u.id = ug.user_id
WHERE ug.group_id = ?
''', (group2_id,))
for row in cur.fetchall():
    print(f"Group Users has member: {row[0]}")

conn.close()

Step4：いろいろなクエリ操作

JOIN/FILTER などのクエリ操作です．

SQLAlchemy

from sqlalchemy import Column, Integer, String, create_engine, ForeignKey, func, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()
db_path = "mydb.db"

"""
Model Definition
"""

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    posts = relationship("Post", back_populates="user")

class Post(Base):
    __tablename__ = 'posts'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    user_id = Column(Integer, ForeignKey('users.id'))
    user = relationship("User", back_populates="posts")

"""
Engine and Table Creation
"""

engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.drop_all(engine)
Base.metadata.create_all(engine)

"""
Session Creation 
"""

Session = sessionmaker(bind=engine)
session = Session()

# ユーザー作成
alice = User(name="Alice")
bob = User(name="Bob")
carol = User(name="Carol")

# 投稿作成
alice.posts = [
    Post(title="Alice's first post"),
    Post(title="Alice's second post"),
]

bob.posts = [
    Post(title="Bob's only post"),
]

carol.posts = []  # 投稿なしのユーザーも作る！

# 登録
session.add_all([alice, bob, carol])
session.commit()

print("*"*10, "Show All", "*"*10)
results = session.query(User).all()
for result in results:
    posts = session.query(Post).filter_by(user_id=result.id)
    print(f"Name: {result.name}")
    for post in posts:
        print(f"\tTitle: {post.title}")
print()


# JOIN
print("*" * 10, "JOIN: Post Title with User Name", "*" * 10)
results = session.query(Post.title, User.name).join(User).all()
for title, username in results:
    print(f"{title} by {username}")

# FILTER
print("*" * 10, "FILTER: Alice or Bob's posts", "*" * 10)
posts = session.query(Post).join(User).filter(or_(
    User.name == "Alice",
    User.name == "Bob",
)).all()
for post in posts:
    print(f"{post.title} (user_id={post.user_id})")

# LIKE
print("*" * 10, "LIKE: Posts with 'second'", "*" * 10)
posts = session.query(Post).filter(Post.title.ilike("%second%")).all()
for post in posts:
    print(post.title)

# ORDER BY
print("*" * 10, "ORDER BY: Posts Desc", "*" * 10)
posts = session.query(Post).order_by(Post.id.desc()).all()
for post in posts:
    print(f"{post.id}: {post.title}")


# GROPU BY
print("*" * 10, "GROUP BY: Post Count per User", "*" * 10)
results = session.query(User.name, func.count(Post.id)) \
    .join(Post) \
    .group_by(User.id) \
    .order_by(func.count(Post.id).desc()) \
    .all()

for name, count in results:
    print(f"{name} has {count} posts")

# COUNT/EXISTS
print("*" * 10, "COUNT / EXISTS", "*" * 10)
# 総投稿数
post_count = session.query(Post).count()
print("Total posts:", post_count)

# Carolに投稿があるか？
has_post = session.query(Post).join(User)\
    .filter(User.name == "Carol")\
    .first() is not None
print("Carol has posts?", has_post)

SQLite3

import sqlite3

conn = sqlite3.connect("sample_raw.db")
cur = conn.cursor()

# 初期化：テーブル削除（必要なら）
cur.execute("DROP TABLE IF EXISTS posts")
cur.execute("DROP TABLE IF EXISTS users")

# テーブル作成
cur.execute('''
CREATE TABLE users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT
)
''')

cur.execute('''
CREATE TABLE posts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT,
    user_id INTEGER,
    FOREIGN KEY(user_id) REFERENCES users(id)
)
''')

# ユーザー作成
cur.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
alice_id = cur.lastrowid

cur.execute("INSERT INTO users (name) VALUES (?)", ("Bob",))
bob_id = cur.lastrowid

cur.execute("INSERT INTO users (name) VALUES (?)", ("Carol",))
carol_id = cur.lastrowid

# 投稿作成
cur.executemany("INSERT INTO posts (title, user_id) VALUES (?, ?)", [
    ("Alice's first post", alice_id),
    ("Alice's second post", alice_id),
    ("Bob's only post", bob_id)
])

conn.commit()

print("*" * 10, "Show All", "*" * 10)
cur.execute("SELECT id, name FROM users")
for user_id, name in cur.fetchall():
    print(f"Name: {name}")
    cur.execute("SELECT title FROM posts WHERE user_id = ?", (user_id,))
    for (title,) in cur.fetchall():
        print(f"\tTitle: {title}")

print("*" * 10, "JOIN: Post Title with User Name", "*" * 10)
cur.execute('''
SELECT posts.title, users.name
FROM posts
JOIN users ON posts.user_id = users.id
''')
for title, username in cur.fetchall():
    print(f"{title} by {username}")

print("*" * 10, "FILTER: Alice or Bob's posts", "*" * 10)
cur.execute('''
SELECT posts.title, posts.user_id
FROM posts
JOIN users ON posts.user_id = users.id
WHERE users.name = ? OR users.name = ?
''', ("Alice", "Bob"))
for title, uid in cur.fetchall():
    print(f"{title} (user_id={uid})")

print("*" * 10, "LIKE: Posts with 'second'", "*" * 10)
cur.execute("SELECT title FROM posts WHERE title LIKE ?", ("%second%",))
for (title,) in cur.fetchall():
    print(title)

print("*" * 10, "ORDER BY: Posts Desc", "*" * 10)
cur.execute("SELECT id, title FROM posts ORDER BY id DESC")
for pid, title in cur.fetchall():
    print(f"{pid}: {title}")

print("*" * 10, "GROUP BY: Post Count per User", "*" * 10)
cur.execute('''
SELECT users.name, COUNT(posts.id)
FROM users
LEFT JOIN posts ON users.id = posts.user_id
GROUP BY users.id
ORDER BY COUNT(posts.id) DESC
''')
for name, count in cur.fetchall():
    print(f"{name} has {count} posts")

print("*" * 10, "COUNT / EXISTS", "*" * 10)

# 総投稿数
cur.execute("SELECT COUNT(*) FROM posts")
print("Total posts:", cur.fetchone()[0])

# Carol に投稿があるか？
cur.execute('''
SELECT 1
FROM posts
JOIN users ON posts.user_id = users.id
WHERE users.name = ?
LIMIT 1
''', ("Carol",))
has_post = cur.fetchone() is not None
print("Carol has posts?", has_post)

conn.close()

最後に

今回は DB 操作に関して，SQLite3 と SQLAlchemy で同じことをしてみました．
複雑なことをしようとすると，SQLAlchemy の方が書きやすい，読みやすいかもしれませんね．

0
1
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
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?