1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- トランザクションA
BEGIN;
UPDATE users SET name = 'Bob' WHERE id = 1;
-- まだ COMMIT していない！

-- トランザクションB
SELECT * FROM users WHERE id = 1;
-- READ UNCOMMITTED なら 'Bob' が見えてしまう（ダーティーリード発生）

-- トランザクションA
ROLLBACK;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

-- トランザクションA
BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT * FROM users WHERE id = 1;  -- 'Alice'

-- トランザクションB
BEGIN;
UPDATE users SET name = 'Bob' WHERE id = 1;
COMMIT;

-- トランザクションA
SELECT * FROM users WHERE id = 1;  -- 'Bob'（前回と結果が違う！）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

-- トランザクションA
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT * FROM users WHERE age >= 18;
-- 3件取得: Alice, Bob, Charlie

-- トランザクションB
BEGIN;
INSERT INTO users (id, name, age) VALUES (4, 'Dave', 22);
COMMIT;

-- トランザクションA
SELECT * FROM users WHERE age >= 18;
-- 4件取得: Alice, Bob, Charlie, Dave（行が増えた！）

1
2
3
4
5
6

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL
);

INSERT INTO users (name) VALUES ('Alice'), ('Bob');

1
2
3
4
5

CREATE TABLE BATCH_LOCK (
    BATCH_ID VARCHAR(50) PRIMARY KEY,  -- バッチ処理の識別子
    LOCK_STATUS CHAR(1) NOT NULL,      -- ロック状態（'1'=ロック中、'0'=未ロック）
    LOCK_TIME TIMESTAMP,               -- ロックを取得した時間
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

BEGIN;

UPDATE BATCH_LOCK 
SET LOCK_STATUS = '1', 
    LOCK_TIME = CURRENT_TIMESTAMP 
WHERE BATCH_ID = 'DAILY_BATCH' 
  AND LOCK_STATUS = '0';

-- 更新件数を確認
GET DIAGNOSTICS affected_rows = ROW_COUNT;
-- affected_rows が 0 ならロック取得失敗

COMMIT;

1
2
3
4
5
6
7
8

BEGIN;

UPDATE BATCH_LOCK 
SET LOCK_STATUS = '0', 
    LOCK_TIME = CURRENT_TIMESTAMP 
WHERE BATCH_ID = 'DAILY_BATCH';

COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

BEGIN;

-- 排他ロックを取得
SELECT LOCK_STATUS, LOCK_TIME 
FROM BATCH_LOCK 
WHERE BATCH_ID = 'DAILY_BATCH' AND LOCK_STATUS = '0' 
FOR UPDATE;

-- 結果の行数をチェック（アプリケーションコード側で）
-- 結果が0行なら、すでにロックされている
-- 結果が1行なら、ロック取得成功

-- ロックされていなければ更新
UPDATE BATCH_LOCK 
SET LOCK_STATUS = '1', 
    LOCK_TIME = CURRENT_TIMESTAMP 
WHERE BATCH_ID = 'DAILY_BATCH' 
  AND LOCK_STATUS = '0';

COMMIT;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

import logging
import json
import sys
from uvicorn.logging import DefaultFormatter

def is_output_a_terminal(stream):
    """
    Check if the provided stream is a terminal.
    """
    return stream.isatty()

class AppJSONFormatter(logging.Formatter):
    def format(self, record):
        full_path = record.pathname
        path_parts = full_path.split("/")

        try:
            my_namespace_index = path_parts.index("my_namespace")
            display_path = "/".join(path_parts[my_namespace_index:])
        except ValueError:
            if len(path_parts) > 3:
                display_path = "/".join(path_parts[-4:])
            else:
                display_path = full_path

        # デフォルトのログフィールド
        log_record = {
            "level": record.levelname,
            "timestamp": self.formatTime(record, "%Y-%m-%d %H:%M:%S"),
            "file": display_path,
            "line": record.lineno,
            "message": record.getMessage(),
        }

        # extra で渡されたカスタムフィールドを追加
        extra_fields = {k: v for k, v in record.__dict__.items() if k not in logging.LogRecord.__dict__}
        log_record.update(extra_fields)

        return json.dumps(log_record, ensure_ascii=False)

use_colors = is_output_a_terminal(sys.stdout)

handler = logging.StreamHandler(sys.stdout)
formatter = AppJSONFormatter()
handler.setFormatter(formatter)

logger = logging.getLogger("app")
logger.setLevel(logging.INFO)
logger.addHandler(handler)

1
2
3
4

import logging
logger = logging.getLogger("app")

logger.info("ユーザーがログインしました", extra={"UserID": 12345, "SessionID": "abcde12345"})

1
2
3
4
5
6
7
8
9

{
    "level": "INFO",
    "timestamp": "2025-02-14 12:34:56",
    "file": "my_namespace/module/auth.py",
    "line": 42,
    "message": "ユーザーがログインしました",
    "UserID": 12345,
    "SessionID": "abcde12345
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

session = Session()

# SELECT文の実行（この時点ではトランザクションは開始されない）
user = session.query(User).first()

# Insertなので、自動的に最初のトランザクションが開始
user1 = User(name="田中一郎")
session.add(user1)
session.commit()  # トランザクション1が終了し、自動的に新しいトランザクションが開始

# Auto Beginで自動的に開始された次のトランザクション
user2 = User(name="鈴木二郎")
session.add(user2)
session.commit()  # トランザクション2が終了し、また新しいトランザクションが開始

# withによる明示的なトランザクション
with session.begin():
    user = User(name="山田太郎")
    session.add(user)
    # withブロックを抜けると自動的にcommitされる

# または
session.begin()
try:
    user = User(name="山田太郎")
    session.add(user)
    session.commit()
except:
    session.rollback()
    raise

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

def get_db():
    db = SessionLocal()
    try:
        yield db  # NOTE: この時点ではトランザクションは開始されていない
    finally:
        db.close()

@app.post("/users/")
def create_user(user: UserCreate, db: Session = Depends(get_db)):
    db_user = User(name=user.name)  
    db.add(db_user)  # この時点で自動的にトランザクションが開始される
    db.commit()  # トランザクションをコミット

@app.get("/users/{user_id}")
def read_user(user_id: int, db: Session = Depends(get_db)):
    # SELECT操作のみなのでトランザクションは開始されない
    user = db.query(User).filter(User.id == user_id).first()
    return user

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy import text

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)

# Async Engine & Session Setup
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)

async def select_for_update_example(user_id: int):
    async with AsyncSessionLocal() as session:
        async with session.begin():  # トランザクション開始
            try:
                # `SELECT ... FOR UPDATE` で行をロック
                stmt = select(User).where(User.id == user_id).with_for_update()
                result = await session.execute(stmt)
                user = result.scalars().first()

                if user:
                    print(f"Locked User: {user.name}")
                    # ここで更新処理などを行う
                    user.name = "Updated Name"
                    await session.commit()
                else:
                    print("User not found")

            except SQLAlchemyError as e:
                await session.rollback()
                print(f"Error: {e}")

1
2

import asyncio
asyncio.run(select_for_update_example(1))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

SELECT client_addr, client_port, query
FROM pg_stat_activity
WHERE state = 'active'
  AND wait_event_type = 'Lock'
ORDER BY state_change;

  client_addr  │ client_port │                       query
═══════════════╪═════════════╪════════════════════════════════════════════════════
 192.168.2.120 │       49578 │ SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
 192.168.2.122 │       47432 │ SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
(2 rows)

1
2

$ lsof -P | grep 49578
myapp  3974  mypc  [...]  TCP localhost:49578->dbserver:5432 (ESTABLISHED)

1
2
3

SELECT pid, relation::regclass, mode, granted
FROM pg_locks
WHERE relation = 'users'::regclass;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

mydb=> SHOW deadlock_timeout;
 deadlock_timeout
------------------
 1s
(1 row)

mydb=> SHOW log_lock_waits;
 log_lock_waits
----------------
 off
(1 row)

1
2

log_statement = 'all'
log_line_prefix = '%m [pid=%p] %q[txid=%x] %u@%d '

1
2

ALTER SYSTEM SET log_lock_waits = ON;
SELECT pg_reload_conf();

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

2022-06-20 17:17:50.888 CEST [pid=19718] [txid=0] mypc@test LOG:  statement: BEGIN;
2022-06-20 17:17:52.988 CEST [pid=19718] [txid=0] mypc@test LOG:  statement: INSERT INTO child VALUES (100, 1, 'new child');
2022-06-20 17:17:58.215 CEST [pid=19674] [txid=0] mypc@test LOG:  statement: BEGIN;
2022-06-20 17:18:03.626 CEST [pid=19674] [txid=0] mypc@test LOG:  statement: INSERT INTO child VALUES (101, 1, 'another child');
2022-06-20 17:18:18.979 CEST [pid=19674] [txid=1108] mypc@test LOG:  statement: SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
2022-06-20 17:18:31.550 CEST [pid=19718] [txid=1109] mypc@test LOG:  statement: SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
2022-06-20 17:18:32.551 CEST [pid=19718] [txid=1109] mypc@test ERROR:  deadlock detected
2022-06-20 17:18:32.551 CEST [pid=19718] [txid=1109] mypc@test DETAIL:  Process 19718 waits for ShareLock on transaction 1108; blocked by process 19674.
        Process 19674 waits for ShareLock on transaction 1109; blocked by process 19718.
        Process 19718: SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
        Process 19674: SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;
2022-06-20 17:18:32.551 CEST [pid=19718] [txid=1109] mypc@test HINT:  See server log for query details.
2022-06-20 17:18:32.551 CEST [pid=19718] [txid=1109] mypc@test CONTEXT:  while locking tuple (0,1) in relation 'parent'
2022-06-20 17:18:32.551 CEST [pid=19718] [txid=1109] mypc@test STATEMENT:  SELECT pdata FROM parent WHERE pid = 1 FOR UPDATE;

1

shared_preload_libraries = 'pg_stat_statements'

1

$ sudo systemctl restart postgresql

1

CREATE EXTENSION pg_stat_statements;

1
2
3
4
5

SELECT query, calls, total_exec_time, rows
FROM pg_stat_statements
WHERE query LIKE '%analysis%'
ORDER BY total_exec_time DESC
LIMIT 10;