{%hackmd GJ0m4S6vQDSgWuvdxcgpGw %} # Jp. SQLAlchemy [TOC] :::spoiler 第三方模組 ```toml= [project] name = "playground" version = "0.1.0" description = "fastapi[all] has rich and pydantic, psycopg[binary] has tzdata." readme = "README.md" requires-python = ">=3.13" dependencies = [ "fastapi[all]>=0.116.1", "psycopg[binary]>=3.2.9", "sqlalchemy>=2.0.42", ] ``` ::: ## 1. PostgreSQL - Roadmap | Group | Description | Keywords | | ------- | ----------------------------------- | ------------------------------------------- | | ==DDL== | 資料庫、資料表、約束、索引、型別 | `NULL` `DEFAULT` `PK` `UNIQUE` `CHECK` `FK` | | DML | 資料操作語言 | `INSERT` `UPDATE` `DELETE` `FOR UPDATE` | | TCL | `START TRANSACTION` | `BEGIN` `COMMIT` `ROLLBACK` | | ==DQL== | 運算子、SQL 函式、關聯式查詢 | `SELECT` `FROM` `JOIN` `WHERE` | | | 聚合函式、統計函式 | `GROUP BY` `HAVING` | | | `ASC NULLS LAST` `DESC NULLS FIRST` | `ORDER BY` `LIMIT` `OFFSET` | | ==DCL== | 資料控制語言 | `GRANT` `REVOKE` | :::spoiler Cheatsheet - 使用者、資料庫、權限 ```sql= -- ========================================================================= -- 建立使用者: CREATE USER CREATE USER super_user WITH SUPERUSER PASSWORD 'super_password'; CREATE USER app_user WITH PASSWORD 'app_password'; -- 刪除使用者: DROP USER DROP USER super_user; DROP USER app_user; -- ========================================================================= -- 建立資料庫: CREATE DATABASE CREATE DATABASE example OWNER super_user; -- 刪除資料庫: DROP DATABASE DROP DATABASE example; -- ========================================================================= -- (example) 授予權限: GRANT TO (super_user → app_user) SET ROLE super_user; GRANT USAGE ON SCHEMA public TO app_user; -- ::: 授予所有表預設權限 ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user; -- ::: 授予現有表權限 GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_user; -- (example) 撤銷權限: REVOKE FROM (super_user → app_user) SET ROLE super_user; REVOKE ALL ON SCHEMA public FROM app_user; -- ::: 撤銷所有表預設權限 ALTER DEFAULT PRIVILEGES IN SCHEMA public REVOKE ALL ON TABLES FROM app_user; -- ::: 撤銷現有表權限 REVOKE ALL ON ALL TABLES IN SCHEMA public FROM app_user; -- ========================================================================= -- 額外補充: MySQL 權限相關語法 CREATE USER 'super_user'@'localhost' IDENTIFIED BY 'super_password'; GRANT ALL ON *.* TO 'super_user'@'localhost' WITH GRANT OPTION; CREATE USER 'app_user'@'localhost' IDENTIFIED BY 'app_password'; GRANT SELECT, INSERT, UPDATE, DELETE ON example.* TO 'app_user'@'localhost'; ``` ::: :::spoiler Cheatsheet - 修改資料表 ```sql= -- ========================================================================= -- DDL CREATE TABLE <tb> (...); -- 建立資料表 DROP TABLE <tb>; -- 刪除資料表 ALTER TABLE <tb> RENAME TO <new_name>; -- 重新命名資料表 -- ========================================================================= -- ALTER Example 1 ALTER TABLE <tb> ALTER <col_name> TYPE <new_type>; -- 變更欄位型別 ALTER TABLE <tb> ALTER <col_name> SET NOT NULL; -- 設定 NOT NULL ALTER TABLE <tb> ALTER <col_name> DROP NOT NULL; -- 刪除 NOT NULL ALTER TABLE <tb> ALTER <col_name> SET DEFAULT <val>; -- 設定 DEFAULT ALTER TABLE <tb> ALTER <col_name> DROP DEFAULT; -- 刪除 DEFAULT -- ALTER Example 2: 新增欄位時，現有的列的值會被設定成 NULL 或約束的預設值 ALTER TABLE <tb> ADD <col_name> <col_type> <col_constraint>; -- 新增欄位 ALTER TABLE <tb> DROP <col_name>; -- 刪除欄位 ALTER TABLE <tb> RENAME <col_name> TO <new_name>; -- 重新命名欄位 -- ========================================================================= -- CONSTRAINT Example 1 ALTER TABLE <tb> ADD CONSTRAINT <constraint_name> <body>; -- 新增約束 ALTER TABLE <tb> DROP CONSTRAINT <constraint_name>; -- 刪除約束 -- CONSTRAINT Example 2 ALTER TABLE <tb> ADD PRIMARY KEY (id); ALTER TABLE <tb> ADD UNIQUE (name); ALTER TABLE <tb> ADD CHECK (name ~ '(?i)haru'); ALTER TABLE <tb> ADD FOREIGN KEY (team_id) REFERENCES team (id); -- NULLS by default distinct, but regarded as equal if NULLS NOT DISTINCT. ALTER TABLE <tb> ADD UNIQUE NULLS NOT DISTINCT (name); ``` ::: :::spoiler Cheatsheet - 備份資料、清除資料、還原資料 ```sql= -- 備份資料 CREATE SCHEMA backup AUTHORIZATION super_user; CREATE TABLE backup.shipments_backup AS TABLE public.shipments; -- 清除資料 (僅在使用 Identity 時需要設定 setval()) TRUNCATE shipments; SELECT setval(pg_get_serial_sequence('shipments', 'id'), 1, false); -- 還原資料 (僅在使用 Identity 時需要設定 setval()) SELECT setval( pg_get_serial_sequence('shipments', 'id'), (SELECT max(id) FROM shipments) ); INSERT INTO shipments SELECT * FROM backup.shipments_backup; ``` ::: :::spoiler Cheatsheet - TRIGGER、pgAdmin4 ```sql= -- 建立觸發器函式 CREATE FUNCTION before_update_set_timestamp() RETURNS TRIGGER AS $$ BEGIN NEW.updated_at := CURRENT_TIMESTAMP; RETURN NEW; END; $$ LANGUAGE plpgsql; -- 建立觸發器 CREATE TRIGGER shipments_update_timestamp BEFORE UPDATE ON shipments FOR EACH ROW EXECUTE FUNCTION before_update_set_timestamp(); -- pgAdmin4 建議設定 -- Preferences → Query Tool → Auto completion: Autocomplete on key press -- Preferences → Query Tool → SQL formatting: Use spaces ``` ::: ## 2. PostgreSQL - Concept ### 2-1. 資料庫正規化 以降低資料的重複性來避免資料更新異常，進而提升資料庫的效率與維護性。 :::spoiler 資料庫正規化: 1NF, 2NF, 3NF, BCNF 1. **第一正規化 (1NF)** - 每一列資料的欄位值都只能是單一值。 - 沒有任何兩筆以上的資料完全重複。 - 資料表中需要有主鍵，其他所有欄位都相依於主鍵。 2. **第二正規化 (2NF)** - 符合 1NF 且沒有部份相依，每一個非主鍵欄位必須完全相依主鍵。 - 僅在主鍵有多個欄位組成時會發生部份相依。 3. **第三正規化 (3NF)** - 符合 2NF 且沒有遞移相依，每一個非主鍵欄位必須直接相依主鍵。 - 非主鍵欄位彼此之間應該要具備獨立性。 4. **Boyce-Codd Normal Form (BCNF)** - 符合 3NF 且多欄位組成的主鍵，各欄位不可以相依於其他非主鍵欄位。 - 僅在主鍵有多個欄位組成時會發生符合 3NF 卻不符合 BCNF 的情況。 ::: ### 2-2. ACID 特性 資料庫管理系統在寫入或更新資料的過程中，為保證交易是正確可靠的，所必須具備的四個特性。 :::spoiler ACID: Atomicity, Consistency, Isolation, Durability 1. **原子性 (Atomicity)** - 交易必須全部成功，否則必須全部回滾到交易之前的狀態。 2. **一致性 (Consistency)** - 交易前後資料庫的完整性沒有被破壞，資料必須符合所有約束、觸發器、級聯回滾等。 3. **隔離性 (Isolation)** - 防止多個交易並行時交叉執行，導致資料不一致的狀況。 4. **持久性 (Durability)** - 交易成功後，資料的變更是永久的，即使系統故障也不會丟失資料。 ::: ### 2-3. ORM 定義 ORM (Object Relational Mapping) 是一種軟體開發模式，允許應用程式以物件導向的方式來操作關聯式資料庫。 ORM 的優點有高抽象性、高安全性、跨資料庫的移植性等，缺點則是可能會有效能問題、以及對於複雜的查詢會有維護性低的問題。 ## 3. SQLAlchemy - Models & Session SQL 主要型別包含字串類型、數值類型、日期類型、布林類型和 JSON 五大種類: :::spoiler SQL - DataTypes ```sql= -- TEXT (equivalent to): VARCHAR -- SMALLINT (2 bytes): 32767 -- INTEGER (4 bytes): 2147483647 -- BIGINT (8 bytes): 9223372036854775807 -- AUTOINCREMENT (....): GENERATED ALWAYS AS IDENTITY CREATE TABLE IF NOT EXISTS characters ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT NOT NULL UNIQUE, age INTEGER NOT NULL CHECK (age >= 0), birth DATE NOT NULL, deposit NUMERIC NOT NULL, is_kawaii BOOLEAN NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, deleted_at TIMESTAMPTZ ); CREATE UNIQUE INDEX ix_characters_lower_name ON characters (lower(name)); CREATE INDEX ix_characters_created_at ON characters (created_at); -- 建立索引的參考準則: -- PK 和 UNIQUE 會自動建立索引。 -- FK 和經常用來查詢、分組或排序的欄位，適合建立索引。 -- 值分布非常廣泛不重複的 NOT NULL 欄位，也適合建立索引。 CREATE TABLE IF NOT EXISTS shipments ( id UUID PRIMARY KEY DEFAULT gen_random_uuid(), name TEXT NOT NULL, price INTEGER NOT NULL, detail JSONB CHECK (jsonb_typeof(detail) = 'object'), character_id UUID NOT NULL REFERENCES characters (id), created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, deleted_at TIMESTAMPTZ ); CREATE INDEX ix_shipments_detail ON shipments USING gin (detail); CREATE INDEX ix_shipments_character_id ON shipments (character_id); CREATE INDEX ix_shipments_created_at ON shipments (created_at); -- scalar: 純量 (single column or model) -- await session.scalar() >>> Execute & return scalar result (single row) -- await session.scalars() >>> Execute & return scalar results (multiple rows) -- >>> all() >>> Fetch all -- >>> first() >>> Fetch the first one or None -- >>> one_or_none() >>> Fetch at most one (2↑: raise error, 0: None) -- >>> one() >>> Fetch exactly one (2↑: raise error, 0: raise error) ``` ::: :::spoiler SQLAlchemy - models ```python= from __future__ import annotations from datetime import date, datetime from decimal import Decimal from uuid import UUID from sqlalchemy import CheckConstraint, DateTime, ForeignKey, Index, text from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import ( DeclarativeBase, Mapped, declared_attr, mapped_column, relationship, ) from sqlalchemy.sql.functions import current_timestamp class Base(DeclarativeBase): __mapper_args__ = {'confirm_deleted_rows': False} @declared_attr def created_at(cls) -> Mapped[datetime]: return mapped_column( DateTime(timezone=True), server_default=current_timestamp(), index=True, ) @declared_attr def updated_at(cls) -> Mapped[datetime]: return mapped_column( DateTime(timezone=True), server_default=current_timestamp(), ) @declared_attr def deleted_at(cls) -> Mapped[datetime | None]: return mapped_column(DateTime(timezone=True)) class Character(Base): __tablename__ = 'characters' __table_args__ = ( # Case insensitive unique index for name Index('ix_characters_lower_name', text('lower(name)'), unique=True), ) id: Mapped[UUID] = mapped_column( primary_key=True, server_default=text('gen_random_uuid()'), ) name: Mapped[str] = mapped_column(unique=True) age: Mapped[int] = mapped_column(CheckConstraint('age >= 0')) birth: Mapped[date] deposit: Mapped[Decimal] is_kawaii: Mapped[bool] shipments: Mapped[list[Shipment]] = relationship( back_populates='character', lazy='selectin', ) def __repr__(self) -> str: return f"Character(id='{self.id}', name='{self.name}')" class Shipment(Base): __tablename__ = 'shipments' __table_args__ = ( # Gin index for details Index('ix_shipments_detail', text('detail'), postgresql_using='gin'), ) id: Mapped[UUID] = mapped_column( primary_key=True, server_default=text('gen_random_uuid()'), ) name: Mapped[str] price: Mapped[int] detail: Mapped[dict | None] = mapped_column( JSONB(none_as_null=True), CheckConstraint("jsonb_typeof(detail) = 'object'"), ) character_id: Mapped[UUID] = mapped_column( ForeignKey(Character.id), index=True, ) character: Mapped[Character] = relationship( back_populates='shipments', lazy='selectin', ) def __repr__(self) -> str: return f"Shipment(id='{self.id}', name='{self.name}')" ``` <p style="margin-top: -1.5rem; margin-left: 0.75rem; opacity: 0.5;">app/db/models.py</p> ::: :::spoiler SQLAlchemy - session ```python= import os from dotenv import load_dotenv from sqlalchemy import delete from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.db.models import Base, Character, Shipment load_dotenv() # DATABASE_URL="postgresql+psycopg://postgres:password@localhost/example" database_url = os.getenv('DATABASE_URL', '') # Recommend for sqlite engine: connect_args={'check_same_thread': False} engine = create_async_engine(database_url) Session = async_sessionmaker(engine, autoflush=False, expire_on_commit=False) async def init_db(reset: bool = False) -> None: async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) if reset: async with Session() as session: await session.execute(delete(Shipment)) await session.execute(delete(Character)) await session.commit() ``` <p style="margin-top: -1.5rem; margin-left: 0.75rem; opacity: 0.5;">app/db/session.py</p> ::: ## 4. SQLAlchemy - Execution 物件層操作在查詢時不會上鎖資料庫，對於資料庫的壓力比較小，並且容易處理複雜的流程，適合使用在簡單的 RESTful API 單筆資料的 CRUD，有不錯的維護性和安全性。 直接使用 SQL 語法會在 UPDATE 和 DELETE 執行時鎖定資料列，容易長時間被占用，但由於擁有比較高的效率和性能，因此適合使用在批量操作上。 :::spoiler SQLAlchemy - ORM Operation ```python= import asyncio from datetime import UTC, date, datetime from rich import print from sqlalchemy import func, select from app.db.models import Character from app.db.session import Session, init_db characters = [ {'name': 'Haru', 'age': 15, 'birth': date(2025, 3, 21)}, {'name': 'Kurea', 'age': 18, 'birth': date(2025, 11, 29)}, {'name': 'Mako', 'age': 19, 'birth': date(2025, 4, 23)}, {'name': 'Shinon', 'age': 19, 'birth': date(2025, 7, 1)}, {'name': 'Hika', 'age': 18, 'birth': date(2025, 9, 9)}, ] deposit_datas = [ {'chara_name': 'Haru', 'deposit': 1_200_000}, {'chara_name': 'Kurea', 'deposit': 1_300_000}, {'chara_name': 'Mako', 'deposit': 1_100_000}, {'chara_name': 'Shinon', 'deposit': 800_000}, {'chara_name': 'Hika', 'deposit': 0}, ] async def main(): await init_db(reset=True) async with Session() as session: for data in characters: character = Character(**data, deposit=0, is_kawaii=True) session.add(character) await session.commit() await session.refresh(character) print(f'Insert: {character}') async with Session() as session: for data in deposit_datas: stmt = ( select(Character) .where( Character.name == data['chara_name'], ) .with_for_update() ) if character := await session.scalar(stmt): character.deposit = data['deposit'] character.updated_at = datetime.now(UTC) await session.commit() await session.refresh(character) print(f'Update: {character.name} -> {character.deposit}') async with Session() as session: stmt = select(Character).where(Character.deposit == 0) for character in await session.scalars(stmt): await session.delete(character) await session.commit() print(f'Delete: {character}') async with Session() as session: stmt = select(func.count()).select_from(Character) print(await session.scalar(stmt)) stmt = select(Character).order_by(Character.created_at) for character in await session.scalars(stmt): print(f'>>> {character}') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # Insert: Character(id='fdf14ce8-7f60-4d52-ac41-087dfc6307ad', name='Haru') # Insert: Character(id='a0add3fe-d6b3-49e2-accd-6b7e6cd35f77', name='Kurea') # Insert: Character(id='bb9e736d-93b9-4654-9b1f-b226ba048d5a', name='Mako') # Insert: Character(id='df34ecd3-abf6-49ed-a9d1-f827fb8ba666', name='Shinon') # Insert: Character(id='e27f27ad-74dd-41cb-87cc-42ac9188c079', name='Hika') # Update: Haru -> 1200000 # Update: Kurea -> 1300000 # Update: Mako -> 1100000 # Update: Shinon -> 800000 # Update: Hika -> 0 # Delete: Character(id='e27f27ad-74dd-41cb-87cc-42ac9188c079', name='Hika') # 4 # >>> Character(id='fdf14ce8-7f60-4d52-ac41-087dfc6307ad', name='Haru') # >>> Character(id='a0add3fe-d6b3-49e2-accd-6b7e6cd35f77', name='Kurea') # >>> Character(id='bb9e736d-93b9-4654-9b1f-b226ba048d5a', name='Mako') # >>> Character(id='df34ecd3-abf6-49ed-a9d1-f827fb8ba666', name='Shinon') ``` ::: :::spoiler SQLAlchemy - Bulk Operation ```python= import asyncio from datetime import UTC, date, datetime from rich import print from sqlalchemy import bindparam, delete, func, insert, select, update from app.db.models import Character from app.db.session import Session, init_db characters = [ {'name': 'Haru', 'age': 15, 'birth': date(2025, 3, 21)}, {'name': 'Kurea', 'age': 18, 'birth': date(2025, 11, 29)}, {'name': 'Mako', 'age': 19, 'birth': date(2025, 4, 23)}, {'name': 'Shinon', 'age': 19, 'birth': date(2025, 7, 1)}, {'name': 'Hika', 'age': 18, 'birth': date(2025, 9, 9)}, ] deposit_datas = [ {'chara_name': 'Haru', 'deposit': 1_200_000}, {'chara_name': 'Kurea', 'deposit': 1_300_000}, {'chara_name': 'Mako', 'deposit': 1_100_000}, {'chara_name': 'Shinon', 'deposit': 800_000}, {'chara_name': 'Hika', 'deposit': 0}, ] async def main(): await init_db(reset=True) async with Session() as session: await session.execute( insert(Character).values(deposit=0, is_kawaii=True), characters, ) await session.commit() print('Bulk Insert') async with Session() as session: conn = await session.connection() await conn.execute( update(Character) .where( Character.name == bindparam('chara_name'), ) .values(updated_at=datetime.now(UTC)), deposit_datas, ) await session.commit() print('Bulk Update') async with Session() as session: stmt = ( delete(Character) .where( Character.deposit == 0, ) .returning(Character) ) character = await session.scalar(stmt) await session.commit() print(f'Delete: {character}') async with Session() as session: stmt = select(func.count()).select_from(Character) print(await session.scalar(stmt)) stmt = select(Character).order_by(Character.created_at) for character in await session.scalars(stmt): print(f'>>> {character}') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # Bulk Insert # Bulk Update # Delete: Character(id='e27f27ad-74dd-41cb-87cc-42ac9188c079', name='Hika') # 4 # >>> Character(id='fdf14ce8-7f60-4d52-ac41-087dfc6307ad', name='Haru') # >>> Character(id='a0add3fe-d6b3-49e2-accd-6b7e6cd35f77', name='Kurea') # >>> Character(id='bb9e736d-93b9-4654-9b1f-b226ba048d5a', name='Mako') # >>> Character(id='df34ecd3-abf6-49ed-a9d1-f827fb8ba666', name='Shinon') ``` ::: ### 4-1. SQLAlchemy - Operators :::spoiler 測試資料: data/characters.json ```json= [ { "id": "fdf14ce8-7f60-4d52-ac41-087dfc6307ad", "name": "Haru", "age": 15, "birth": "2025-03-21", "deposit": 1200000, "is_kawaii": true }, { "id": "a0add3fe-d6b3-49e2-accd-6b7e6cd35f77", "name": "Kurea", "age": 18, "birth": "2025-11-29", "deposit": 1300000, "is_kawaii": true }, { "id": "bb9e736d-93b9-4654-9b1f-b226ba048d5a", "name": "Mako", "age": 19, "birth": "2025-04-23", "deposit": 1100000, "is_kawaii": true }, { "id": "df34ecd3-abf6-49ed-a9d1-f827fb8ba666", "name": "Shinon", "age": 19, "birth": "2025-07-01", "deposit": 800000, "is_kawaii": true } ] ``` <p style="margin-top: -1.5rem; margin-left: 0.75rem; opacity: 0.5;">data/characters.json</p> ::: :::spoiler 測試資料: data/shipments.json ```json= [ { "id": "da598202-fb90-4d48-abf0-8cbbeebe5309", "name": "Recipe", "price": 1540, "detail": { "authors": [ { "name": "Benita", "pet": "cat" }, { "name": "Agustina", "pet": "dog" } ] }, "character_id": "fdf14ce8-7f60-4d52-ac41-087dfc6307ad" }, { "id": "4f181ed7-fc19-4f3b-88ee-9a74a3f8bd22", "name": "Hair Ring", "price": 1560, "detail": { "brand": { "name": "Vanilla", "code": 374062 } }, "character_id": "fdf14ce8-7f60-4d52-ac41-087dfc6307ad" }, { "id": "276b467a-7a26-4b3e-949a-d14d3c42f3ba", "name": "Swimsuit", "price": 1610, "detail": { "brand": { "name": "Vanilla", "code": 495380 } }, "character_id": "fdf14ce8-7f60-4d52-ac41-087dfc6307ad" }, { "id": "db65ad26-d69f-4fd1-aeb3-0f73c31db310", "name": "Recipe", "price": 1540, "detail": { "authors": [ { "name": "Benita", "pet": "cat" }, { "name": "Viva", "pet": "fox" } ] }, "character_id": "a0add3fe-d6b3-49e2-accd-6b7e6cd35f77" }, { "id": "1d5db208-5049-4fea-866a-b85c5bd0235b", "name": "Hair Ring", "price": 1560, "detail": { "brand": { "name": "Vanilla", "code": 527614 } }, "character_id": "a0add3fe-d6b3-49e2-accd-6b7e6cd35f77" }, { "id": "c7bce059-0b60-49f4-bc92-94db72637e6c", "name": "Recipe", "price": 1540, "detail": null, "character_id": "bb9e736d-93b9-4654-9b1f-b226ba048d5a" } ] ``` <p style="margin-top: -1.5rem; margin-left: 0.75rem; opacity: 0.5;">data/shipments.json</p> ::: :::spoiler SQL - Operators | Group | SQLAlchemy Syntax | | ---------- | ------------------------------------------------------- | | Arithmetic | `+` `-` `*` `/` `%` | | Comparison | `=` `<` `>` `<=` `>=` `!=` `is_(None)` `is_not(None)` | | Logical | `~` `not_()` `&` `and_()` `\|` `or_()` | | Common | `between()` `in_()` `like()` `ilike()` `regexp_match()` | | Negative | `not_in()` `not_like()` `not_ilike()` | ::: :::spoiler SQLAlchemy - DISTINCT ```python= import asyncio import json from rich import print from sqlalchemy import func, insert, select from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() # Operators: Arithmetic, Comparison, Logical (use bitwise or from import) print('SELECT * FROM characters') print('WHERE deposit >= age * 80000 AND deleted_at IS NULL;') async with Session() as session: stmt = select(Character).where( Character.deposit >= Character.age * 80000, Character.deleted_at.is_(None), ) result = await session.scalars(stmt) print(f'>>> {result.one()}\n') # Operators and extract() function print('SELECT * FROM characters') print('WHERE extract(month from birth) IN (3, 5, 7)') print(' AND extract(day from birth) BETWEEN 21 AND 29;') async with Session() as session: stmt = select(Character).where( func.extract('month', Character.birth).in_([3, 5, 7]), func.extract('day', Character.birth).between(21, 29), ) result = await session.scalars(stmt) print(f'>>> {result.one()}\n') # DISTINCT print('SELECT DISTINCT age FROM characters ORDER BY age DESC;') async with Session() as session: stmt = select(Character.age).distinct().order_by(Character.age.desc()) result = await session.scalars(stmt) print(f'>>> {result.all()}\n') # DISTINCT and count() function # - count(*) : Return quantity # - count(col_name) : Return quantity, ignore NULL # - count(DISTINCT col_name): Return quantity, ignore NULL, distinct print('SELECT count(DISTINCT age) FROM characters;') async with Session() as session: stmt = select(func.count(Character.age.distinct())) print(f'>>> {await session.scalar(stmt)}') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # SELECT * FROM characters # WHERE deposit >= age * 80000 AND deleted_at IS NULL; # >>> Character(id='fdf14ce8-7f60-4d52-ac41-087dfc6307ad', name='Haru') # SELECT * FROM characters # WHERE extract(month from birth) IN (3, 5, 7) # AND extract(day from birth) BETWEEN 21 AND 29; # >>> Character(id='fdf14ce8-7f60-4d52-ac41-087dfc6307ad', name='Haru') # SELECT DISTINCT age FROM characters ORDER BY age DESC; # >>> [19, 18, 15] # SELECT count(DISTINCT age) FROM characters; # >>> 3 ``` ::: ### 4-2. SQLAlchemy - Functions :::spoiler SQL - Functions | Group | SQLAlchemy Syntax | | ----------- | ------------------------------------------------------------------------ | | String | `func.length(x)` `func.upper(x)` `func.lower(x)` `func.replace(x, a, b)` | | | `func.concat(...args)` `func.trim(x[, chars])` | | | `func.substring(x, start, count)` → ==1-based== | | | ==PostgreSQL== `func.split_part(x, sep, n)` → ==1-based== | | | ==PostgreSQL== `func.position(literal(str).op('IN')(x))` → ==1-based== | | Numeric | `func.abs(x)` `func.floor(x)` `func.ceil(x)` `func.round(x, n)` | | | `func.mod(x, y)` `func.pow(x, y)` `func.sqrt(x)` | | | ==PostgreSQL== `func.cbrt(x)` `func.random()` `func.trunc(x, n)` | | Datetime | `func.current_timestamp()` `func.extract(field, x)` | | Nullish | `func.coalesce(...args)` `func.nullif(x, y)` | | Aggregation | `func.count(x)` `func.min(x)` `func.max(x)` `func.sum(x)` `func.avg(x)` | ::: :::spoiler SQLAlchemy - CAST ```python= import asyncio import json from datetime import UTC, datetime, timedelta from rich import print from sqlalchemy import NUMERIC, func, insert, select from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() # String functions print("SELECT name || '-' || age AS concat FROM characters;") print("SELECT concat(name, '-', age) FROM characters;") async with Session() as session: stmt = select(func.concat(Character.name, '-', Character.age)) result = await session.scalars(stmt) print(f'>>> {result.all()!r}\n') # Numeric functions print('SELECT (random() * 100)::numeric(4, 2);') print('SELECT cast(random() * 100 AS numeric(4, 2));') async with Session() as session: stmt = select((func.random() * 100).cast(NUMERIC(4, 2))) print(f'>>> {await session.scalar(stmt)!r}\n') # Datetime functions # - DATE : days between (integer) # - TIMESTAMPTZ: timedelta (interval) print('SELECT created_at - :timestamptz FROM characters;') async with Session() as session: few_days_ago = datetime.now(UTC) - timedelta(days=3, hours=12) stmt = select(Character.created_at - few_days_ago) print(f'>>> {await session.scalar(stmt)!s}\n') # Nullish functions # - coalesce(*x): return first non-NULL value # - nullif(a, b): return a if a != b else NULL print('SELECT coalesce(nullif(3, 3), nullif(21, 0));') async with Session() as session: stmt = select(func.coalesce(func.nullif(3, 3), func.nullif(21, 0))) print(f'>>> {await session.scalar(stmt)!r}\n') # Aggregation functons # - use select_from() if not specified entity in select() async with Session() as session: print('SELECT count(*) FROM characters;') stmt = select(func.count()).select_from(Character) print(f'>>> {await session.scalar(stmt)!r}\n') print('SELECT avg(deposit)::numeric(10) FROM characters;') print('SELECT cast(avg(deposit) AS numeric(10)) FROM characters;') stmt = select(func.avg(Character.deposit).cast(NUMERIC(10))) print(f'>>> {await session.scalar(stmt)!r}') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # SELECT name || '-' || age AS concat FROM characters; # SELECT concat(name, '-', age) FROM characters; # >>> ['Haru-15', 'Kurea-18', 'Mako-19', 'Shinon-19'] # SELECT (random() * 100)::numeric(4, 2); # SELECT cast(random() * 100 AS numeric(4, 2)); # >>> Decimal('27.05') # SELECT created_at - :timestamptz FROM characters; # >>> 3 days, 11:59:59.929602 # SELECT coalesce(nullif(3, 3), nullif(21, 0)); # >>> 21 # SELECT count(*) FROM characters; # >>> 4 # SELECT avg(deposit)::numeric(10) FROM characters; # SELECT cast(avg(deposit) AS numeric(10)) FROM characters; # >>> Decimal('1100000') ``` ::: ### 4-3. SQLAlchemy - Relationship :::spoiler SQLAlchemy - JOIN ```python= import asyncio import json from collections import deque from rich import print from sqlalchemy import insert, select from sqlalchemy.orm import aliased from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() # INNER JOIN print('SELECT * FROM shipments') print(' JOIN characters ON characters.id = shipments.character_id') print("WHERE characters.name = 'Haru'") print('ORDER BY shipments.created_at LIMIT 5 OFFSET 0;') async with Session() as session: stmt = ( select(Shipment) .join(Character) .where(Character.name == 'Haru') .order_by(Shipment.created_at) .offset(0) .limit(5) ) for shipment in await session.scalars(stmt): assert shipment.character.name == 'Haru' print(f'>>> {shipment}') memory = deque[tuple[Character, Shipment | None]]() print() # LEFT OUTER JOIN print('SELECT characters.*, shipments.* FROM characters') print(' LEFT JOIN shipments ON characters.id = shipments.character_id') print('ORDER BY characters.created_at, shipments.created_at;') async with Session() as session: stmt = ( select(Character, Shipment) .join(Shipment, isouter=True) .order_by(Character.created_at, Shipment.created_at) ) for character, shipment in await session.execute(stmt): print(f'>>> {character.name} -> {shipment and shipment.name!r}') memory.appendleft((character, shipment)) # FULL OUTER JOIN # - same result in this example because the relationship constraint async with Session() as session: stmt = ( select(Character, Shipment) .join(Shipment, full=True) .order_by(Character.created_at, Shipment.created_at) ) for character, shipment in await session.execute(stmt): data = memory.pop() assert character.id == data[0].id if data[1] is None: assert shipment is None else: assert shipment.id == data[1].id assert not memory print() # CROSS JOIN alternatives: remove warning message print('SELECT c1.*, c2.* FROM characters c1') print(' JOIN characters c2 ON c2.deposit - c1.deposit = 100000') print('ORDER BY c1.deposit;') async with Session() as session: c1 = aliased(Character) c2 = aliased(Character) stmt = ( select(c1, c2) .join( c2, c2.deposit - c1.deposit == 100000, ) .order_by(c1.deposit) ) character2 = None for character1, target in await session.execute(stmt): print(f'>>> {character1.name} -> {character1.deposit!r}') character2 = target if character2: print(f'$> {character2.name} -> {character2.deposit!r}') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # SELECT * FROM shipments # JOIN characters ON characters.id = shipments.character_id # WHERE characters.name = 'Haru' # ORDER BY shipments.created_at LIMIT 5 OFFSET 0; # >>> Shipment(id='da598202-fb90-4d48-abf0-8cbbeebe5309', name='Recipe') # >>> Shipment(id='4f181ed7-fc19-4f3b-88ee-9a74a3f8bd22', name='Hair Ring') # >>> Shipment(id='276b467a-7a26-4b3e-949a-d14d3c42f3ba', name='Swimsuit') # SELECT characters.*, shipments.* FROM characters # LEFT JOIN shipments ON characters.id = shipments.character_id # ORDER BY characters.created_at, shipments.created_at; # >>> Haru -> 'Recipe' # >>> Haru -> 'Hair Ring' # >>> Haru -> 'Swimsuit' # >>> Kurea -> 'Recipe' # >>> Kurea -> 'Hair Ring' # >>> Mako -> 'Recipe' # >>> Shinon -> None # SELECT c1.*, c2.* FROM characters c1 # JOIN characters c2 ON c2.deposit - c1.deposit = 100000 # ORDER BY c1.deposit; # >>> Mako -> Decimal('1100000') # >>> Haru -> Decimal('1200000') # $> Kurea -> Decimal('1300000') ``` ::: :::spoiler SQLAlchemy - Statistics ```python= import asyncio import json from rich import print from sqlalchemy import NUMERIC, func, insert, select from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() async with Session() as session: stmt = ( select( Character.name, func.count(Shipment.id), func.sum(Shipment.price), func.avg(Shipment.price).cast(NUMERIC(10)), func.percentile_cont(0.5).within_group(Shipment.price), func.stddev_samp(Shipment.price).cast(NUMERIC(4, 2)), func.var_samp(Shipment.price).cast(NUMERIC(8, 4)), func.stddev_pop(Shipment.price).cast(NUMERIC(4, 2)), func.var_pop(Shipment.price).cast(NUMERIC(8, 4)), ) .join(Shipment, isouter=True) .group_by(Character.name) .having(func.count(Shipment.id).is_not(None)) .order_by(func.count(Shipment.id).desc()) ) for character in await session.execute(stmt): print(f'name : {character[0]}') print(f'count : {character[1]}') print(f'sum : {character[2]}') print(f'avg : {character[3]}') print(f'median: {character[4]}') print(f'samp : {character[5]} | {character[6]}') print(f'pop : {character[7]} | {character[8]}') print('----------------------------------------') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # name : Haru # count : 3 # sum : 4710 # avg : 1570 # median: 1560.0 # samp : 36.06 | 1300.0000 # pop : 29.44 | 866.6667 # ---------------------------------------- # name : Kurea # count : 2 # sum : 3100 # avg : 1550 # median: 1550.0 # samp : 14.14 | 200.0000 # pop : 10.00 | 100.0000 # ---------------------------------------- # name : Mako # count : 1 # sum : 1540 # avg : 1540 # median: 1540.0 # samp : None | None # pop : 0.00 | 0.0000 # ---------------------------------------- # name : Shinon # count : 0 # sum : None # avg : None # median: None # samp : None | None # pop : None | None # ---------------------------------------- ``` ::: ### 4-4. SQLAlchemy - Advanced Topics :::spoiler SQLAlchemy - CTE, Window Function ```python= import asyncio import json from rich import print from sqlalchemy import NUMERIC, func, insert, select from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() async with Session() as session: sh_count = func.count(Shipment.id) sh_sum = func.sum(Shipment.price) sh_avg = func.avg(Shipment.price) sh_std = func.stddev_samp(Shipment.price) sh_var = func.var_samp(Shipment.price) cte = ( select( Character.name.label('name'), Character.deposit.label('deposit'), sh_count.over(partition_by=Character.name), sh_sum.over(partition_by=Character.name), sh_avg.over(partition_by=Character.name).cast(NUMERIC(10)), sh_std.over(partition_by=Character.name).cast(NUMERIC(4, 2)), sh_var.over(partition_by=Character.name).cast(NUMERIC(8, 4)), ) .distinct() .join(Shipment, isouter=True) .cte() ) stmt = select( func.rank().over(order_by=cte.c.deposit.desc()).label('rank'), cte, ).order_by('rank') for character in await session.execute(stmt): print(f'rank : {character[0]}') print(f'name : {character[1]}') print(f'deposit: {character[2]}') print(f'count : {character[3]}') print(f'sum : {character[4]}') print(f'avg : {character[5]}') print(f'samp : {character[6]} | {character[7]}') print('----------------------------------------') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # rank : 1 # name : Kurea # deposit: 1300000 # count : 2 # sum : 3100 # avg : 1550 # samp : 14.14 | 200.0000 # ---------------------------------------- # rank : 2 # name : Haru # deposit: 1200000 # count : 3 # sum : 4710 # avg : 1570 # samp : 36.06 | 1300.0000 # ---------------------------------------- # rank : 3 # name : Mako # deposit: 1100000 # count : 1 # sum : 1540 # avg : 1540 # samp : None | None # ---------------------------------------- # rank : 4 # name : Shinon # deposit: 800000 # count : 0 # sum : None # avg : None # samp : None | None # ---------------------------------------- ``` ::: :::spoiler SQLAlchemy - EXISTS ```python= import asyncio import json from rich import print from sqlalchemy import case, insert, select from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() # EXISTS print('SELECT name FROM characters WHERE EXISTS (') print(' SELECT * FROM shipments') print(' WHERE characters.id = shipments.character_id\n);') async with Session() as session: expr = None stmt = select(Character.name).where(Character.shipments.any(expr)) result = await session.scalars(stmt) print(f'>>> {result.all()}\n') # IN (subquery) print('SELECT name FROM characters WHERE id IN (') print(' SELECT character_id FROM shipments\n);') async with Session() as session: subquery = select(Shipment.character_id) stmt = select(Character.name).where(Character.id.in_(subquery)) result = await session.scalars(stmt) print(f'>>> {result.all()}\n') # CASE print('SELECT name, CASE') print(" WHEN age BETWEEN 15 AND 17 THEN 'JK'") print(" WHEN age BETWEEN 18 AND 21 THEN 'JD'") print(" ELSE 'unkwown'") print('END FROM characters;') async with Session() as session: stmt = select( Character.name, case( (Character.age.between(15, 17), 'JK'), (Character.age.between(18, 21), 'JD'), else_='unkwown', ), ) result = await session.execute(stmt) print(f'>>> {result.all()}\n') # UNION ALL print('(SELECT name FROM characters ORDER BY created_at)') print('UNION ALL (SELECT DISTINCT name FROM shipments);') async with Session() as session: stmt1 = select(Character.name).order_by(Character.created_at) stmt2 = select(Shipment.name).distinct() stmt = stmt1.union_all(stmt2) result = await session.scalars(stmt) print(f'>>> {result.all()}\n') asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # SELECT name FROM characters WHERE EXISTS ( # SELECT * FROM shipments # WHERE characters.id = shipments.character_id # ); # >>> ['Haru', 'Kurea', 'Mako'] # SELECT name FROM characters WHERE id IN ( # SELECT character_id FROM shipments # ); # >>> ['Haru', 'Kurea', 'Mako'] # SELECT name, CASE # WHEN age BETWEEN 15 AND 17 THEN 'JK' # WHEN age BETWEEN 18 AND 21 THEN 'JD' # ELSE 'unkwown' # END FROM characters; # >>> [('Haru', 'JK'), ('Kurea', 'JD'), ('Mako', 'JD'), ('Shinon', 'JD')] # (SELECT name FROM characters ORDER BY created_at) # UNION ALL (SELECT DISTINCT name FROM shipments); # >>> ['Haru', 'Kurea', 'Mako', 'Shinon', 'Hair Ring', 'Swimsuit', 'Recipe'] ``` ::: :::spoiler SQLAlchemy - JSONB ```python= import asyncio import json from rich import print from sqlalchemy import insert, select, text from app.db.models import Character, Shipment from app.db.session import Session, init_db with open('data/characters.json') as f1, open('data/shipments.json') as f2: characters = json.load(f1) shipments = json.load(f2) async def main(): await init_db(reset=True) async with Session() as session: if characters: await session.execute(insert(Character), characters) if shipments: await session.execute(insert(Shipment), shipments) await session.commit() # Return JSONB (->) or TEXT (->>) print('SELECT * FROM shipments') print("WHERE detail -> 'brand' ->> 'name' = 'Vanilla';") async with Session() as session: stmt = select(Shipment).from_statement( text(""" SELECT * FROM shipments WHERE detail -> 'brand' ->> 'name' = :name; """).params(name='Vanilla') ) for shipment in await session.scalars(stmt): print(f'>>> {shipment}') print() # Return JSONB (#>) or TEXT (#>>) print('SELECT * FROM shipments') print("WHERE detail #>> '{authors,0,name}' = 'Benita';") async with Session() as session: stmt = select(Shipment).from_statement( text(""" SELECT * FROM shipments WHERE detail #>> '{authors,0,name}' = :name; """).params(name='Benita') ) for shipment in await session.scalars(stmt): print(f'>>> {shipment}') print() # Check if include (@>) print('SELECT * FROM shipments') print('WHERE detail @> \'{"authors":[{"pet": "cat"}]}\';') async with Session() as session: stmt = select(Shipment).from_statement( text(""" SELECT * FROM shipments WHERE detail @> '{"authors":[{"pet": "cat"}]}'; """) ) for shipment in await session.scalars(stmt): print(f'>>> {shipment}') print() # Check if exists by union (?|) or by intersection (?&) print('SELECT * FROM shipments') print("WHERE jsonb_path_query_array(detail, '$.authors.pet')") print(" ?| '{dog,fox}';") async with Session() as session: stmt = select(Shipment).from_statement( text(""" SELECT * FROM shipments WHERE jsonb_path_query_array(detail, '$.authors.pet') ?| '{dog,fox}'; """) ) for shipment in await session.scalars(stmt): print(f'>>> {shipment}') print() # Query by regex print('SELECT * FROM shipments') print('WHERE detail @@ \'$.brand.name like_regex "(?i)^vanilla$"\';') async with Session() as session: stmt = select(Shipment).from_statement( text(""" SELECT * FROM shipments WHERE detail @@ '$.brand.name like_regex "(?i)^vanilla$"'; """) ) for shipment in await session.scalars(stmt): print(f'>>> {shipment}') print() asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop) # 執行結果 (python -m app.main) # ---------------------------------------------------------------------- # SELECT * FROM shipments # WHERE detail -> 'brand' ->> 'name' = 'Vanilla'; # >>> Shipment(id='4f181ed7-fc19-4f3b-88ee-9a74a3f8bd22', name='Hair Ring') # >>> Shipment(id='276b467a-7a26-4b3e-949a-d14d3c42f3ba', name='Swimsuit') # >>> Shipment(id='1d5db208-5049-4fea-866a-b85c5bd0235b', name='Hair Ring') # SELECT * FROM shipments # WHERE detail #>> '{authors,0,name}' = 'Benita'; # >>> Shipment(id='da598202-fb90-4d48-abf0-8cbbeebe5309', name='Recipe') # >>> Shipment(id='db65ad26-d69f-4fd1-aeb3-0f73c31db310', name='Recipe') # SELECT * FROM shipments # WHERE detail @> '{"authors":[{"pet": "cat"}]}'; # >>> Shipment(id='da598202-fb90-4d48-abf0-8cbbeebe5309', name='Recipe') # >>> Shipment(id='db65ad26-d69f-4fd1-aeb3-0f73c31db310', name='Recipe') # SELECT * FROM shipments # WHERE jsonb_path_query_array(detail, '$.authors.pet') # ?| '{dog,fox}'; # >>> Shipment(id='da598202-fb90-4d48-abf0-8cbbeebe5309', name='Recipe') # >>> Shipment(id='db65ad26-d69f-4fd1-aeb3-0f73c31db310', name='Recipe') # SELECT * FROM shipments # WHERE detail @@ '$.brand.name like_regex "(?i)^vanilla$"'; # >>> Shipment(id='4f181ed7-fc19-4f3b-88ee-9a74a3f8bd22', name='Hair Ring') # >>> Shipment(id='276b467a-7a26-4b3e-949a-d14d3c42f3ba', name='Swimsuit') # >>> Shipment(id='1d5db208-5049-4fea-866a-b85c5bd0235b', name='Hair Ring') ``` ::: ::::spoiler 補充 - 關於 self-relationship :::success 1. FK 所對應的 relationship 必須設定: `remote_side=[id]` 2. 查詢時需要主動設定: `select(A).options(selectinload(A.x), selectinload(A.y))` ::: :::: ::::spoiler 補充 - 關於 multiple-foreignkey :::success 1. FK 所對應的 relationship 必須設定: `foreign_keys=[specified_id]` 2. back_populates 對象也必須設定 foreign_keys，但是只能使用字串硬編碼。 ::: ::::