第163回では、AIの要確認結果をSQLiteへ取り込み、担当者が採用・修正・却下を記録するレビューキューを作りました。担当者が複数になると、同じ項目を重複して処理しない仕組みが必要です。
今回は、レビュー項目を一定時間だけ担当者へ割り当てる期限付きロックを実装します。有効期限内は別の担当者による取得を拒否し、失効後は新しい担当者が再取得できます。古いトークンからの確定操作も拒否します。
Python標準ライブラリだけを使い、AIサービスへの通信や元データの更新は行いません。第163回のデータベースへ直接適用する移行スクリプトではないため、動作確認には新しいデータベースを使ってください。
今回実装する契約
| 項目 | 仕様 |
|---|---|
| 取得順 | 未確定かつ未取得、またはロックが失効した項目を、登録時刻、バッチID、レコードIDの順で1件取得する |
| ロック時間 | 30秒以上3600秒以下の整数。現在時刻を整数秒へ切り捨てるため、実際の有効時間は指定値より最大1秒未満短くなり得る |
| 確定条件 | 有効期限内のロックトークンを持つこと |
| 確定結果 | accepted、corrected、rejectedのいずれか |
| 時刻 | time.time()から得たUnix秒を整数で保存する |
| 出力 | ヘルプを除き、標準出力または標準エラー出力へ1行のJSONを返す |
| 対象環境 | 同じ端末上の少数プロセスが、ローカルディスク上のSQLiteを共有する構成 |
reviewerは担当者名の記録であり、本人確認ではありません。トークンも認証やアクセス制御の代わりにはなりません。OSのファイル権限や利用者認証は別に用意します。
期限付きロックの状態変化
| 現在の状態 | 操作 | 結果 |
|---|---|---|
| 未取得 | claim |
担当者、トークン、有効期限を記録する |
| 有効な取得中 | renew |
現在時刻を基準に有効期限を設定し直す |
| 有効な取得中 | release |
確定せずロックを解放する |
| 有効な取得中 | decide |
レビュー結果を確定し、ロック情報を消去する |
| 失効済み | 別の担当者がclaim |
新しいトークンで再取得する |
| 再取得後 | 古いトークンでdecide |
確定を拒否する |
レビュー用CLIを作る
次のコードをreview_leases.pyとして保存します。Python 3.10以上を前提とします。
from __future__ import annotations
import argparse
import json
import secrets
import sqlite3
import sys
import time
from contextlib import closing
from pathlib import Path
from typing import Any
SCHEMA_VERSION = 1
MIN_LEASE_SECONDS = 30
MAX_LEASE_SECONDS = 3600
MAX_CLAIM_FILE_BYTES = 16_384
class AppError(Exception):
def __init__(self, code: str, message: str, exit_code: int = 2):
super().__init__(message)
self.code = code
self.message = message
self.exit_code = exit_code
class JsonArgumentParser(argparse.ArgumentParser):
def error(self, message: str):
raise AppError('argument_error', message)
def emit(payload: dict[str, Any], *, error: bool = False) -> None:
stream = sys.stderr if error else sys.stdout
print(
json.dumps(payload, ensure_ascii=False, separators=(',', ':')),
file=stream,
)
def lease_seconds(value: str) -> int:
try:
seconds = int(value)
except ValueError as exc:
raise argparse.ArgumentTypeError('整数を指定してください') from exc
if not MIN_LEASE_SECONDS <= seconds <= MAX_LEASE_SECONDS:
raise argparse.ArgumentTypeError(
f'{MIN_LEASE_SECONDS}以上{MAX_LEASE_SECONDS}以下で指定してください'
)
return seconds
def validate_text(value: str, field: str, max_length: int) -> str:
if not value or value != value.strip():
raise AppError('invalid_text', f'{field}は前後空白のない文字列にしてください')
if len(value) > max_length:
raise AppError('invalid_text', f'{field}は{max_length}文字以内にしてください')
if any(ord(char) < 32 or ord(char) == 127 for char in value):
raise AppError('invalid_text', f'{field}にASCII制御文字は使用できません')
return value
def validate_token(value: str) -> str:
valid_chars = all(
char.isascii() and (char.isalnum() or char in '-_')
for char in value
)
if not 20 <= len(value) <= 200 or not valid_chars:
raise AppError('invalid_token', 'ロックトークンの形式が正しくありません')
return value
def configure(connection: sqlite3.Connection) -> None:
connection.row_factory = sqlite3.Row
connection.execute('PRAGMA foreign_keys = ON')
connection.execute('PRAGMA busy_timeout = 5000')
def connect_new(path: Path) -> sqlite3.Connection:
if path.exists():
raise AppError('database_exists', '初期化先がすでに存在します')
connection = sqlite3.connect(
str(path), timeout=5.0, isolation_level=None
)
configure(connection)
return connection
def connect_existing(path: Path) -> sqlite3.Connection:
if not path.is_file():
raise AppError('database_not_found', 'データベースが見つかりません')
uri = path.resolve().as_uri() + '?mode=rw'
connection = sqlite3.connect(
uri, uri=True, timeout=5.0, isolation_level=None
)
configure(connection)
return connection
def verify_schema(connection: sqlite3.Connection) -> None:
version = connection.execute('PRAGMA user_version').fetchone()[0]
if version != SCHEMA_VERSION:
raise AppError(
'unsupported_schema',
f'対応していないスキーマです: {version}',
)
def load_claim_token(args: argparse.Namespace) -> str:
if args.token is not None:
return validate_token(args.token)
try:
with args.claim_json.open('rb') as source:
data = source.read(MAX_CLAIM_FILE_BYTES + 1)
except OSError as exc:
raise AppError(
'claim_file_error',
f'取得結果を読めません: {exc}',
) from exc
if len(data) > MAX_CLAIM_FILE_BYTES:
raise AppError('claim_file_error', '取得結果のファイルが大きすぎます')
try:
payload = json.loads(data.decode('utf-8'))
except (UnicodeError, json.JSONDecodeError) as exc:
raise AppError(
'claim_file_error',
f'取得結果がUTF-8のJSONではありません: {exc}',
) from exc
if not isinstance(payload, dict) or not isinstance(
payload.get('claim_token'), str
):
raise AppError(
'claim_file_error',
'取得結果にclaim_tokenがありません',
)
return validate_token(payload['claim_token'])
CREATE_CATEGORY_SQL = '''
CREATE TABLE allowed_category (
category TEXT PRIMARY KEY,
CHECK(length(trim(category)) > 0)
)
'''
CREATE_REVIEW_SQL = '''
CREATE TABLE review_item (
batch_id TEXT NOT NULL,
record_id TEXT NOT NULL,
proposed_category TEXT NOT NULL
REFERENCES allowed_category(category),
reason TEXT NOT NULL,
status TEXT NOT NULL DEFAULT 'pending'
CHECK(status IN ('pending', 'accepted', 'corrected', 'rejected')),
final_category TEXT REFERENCES allowed_category(category),
reviewed_by TEXT,
decided_at INTEGER,
created_at INTEGER NOT NULL,
claim_token TEXT UNIQUE,
claimed_by TEXT,
claimed_at INTEGER,
claim_expires_at INTEGER,
PRIMARY KEY(batch_id, record_id),
CHECK(length(batch_id) BETWEEN 1 AND 100),
CHECK(length(record_id) BETWEEN 1 AND 100),
CHECK(length(reason) BETWEEN 1 AND 1000),
CHECK(
(status = 'pending'
AND final_category IS NULL
AND reviewed_by IS NULL
AND decided_at IS NULL)
OR
(status IN ('accepted', 'corrected')
AND final_category IS NOT NULL
AND reviewed_by IS NOT NULL
AND decided_at IS NOT NULL)
OR
(status = 'rejected'
AND final_category IS NULL
AND reviewed_by IS NOT NULL
AND decided_at IS NOT NULL)
),
CHECK(status <> 'accepted' OR final_category = proposed_category),
CHECK(status <> 'corrected' OR final_category <> proposed_category),
CHECK(
(claim_token IS NULL
AND claimed_by IS NULL
AND claimed_at IS NULL
AND claim_expires_at IS NULL)
OR
(status = 'pending'
AND claim_token IS NOT NULL
AND claimed_by IS NOT NULL
AND claimed_at IS NOT NULL
AND claim_expires_at > claimed_at)
)
)
'''
CREATE_INDEX_SQL = '''
CREATE INDEX review_item_claim_order
ON review_item(status, claim_expires_at, created_at, batch_id, record_id)
'''
def cmd_init(args: argparse.Namespace) -> dict[str, Any]:
categories = [
validate_text(item, 'category', 50) for item in args.category
]
if len(categories) != len(set(categories)):
raise AppError('duplicate_category', 'カテゴリが重複しています')
with closing(connect_new(args.db)) as connection:
with connection:
connection.execute('BEGIN IMMEDIATE')
connection.execute(CREATE_CATEGORY_SQL)
connection.execute(CREATE_REVIEW_SQL)
connection.execute(CREATE_INDEX_SQL)
connection.executemany(
'INSERT INTO allowed_category(category) VALUES (?)',
[(category,) for category in categories],
)
connection.execute(f'PRAGMA user_version = {SCHEMA_VERSION}')
return {
'database': str(args.db),
'category_count': len(categories),
}
def cmd_add(args: argparse.Namespace) -> dict[str, Any]:
batch_id = validate_text(args.batch_id, 'batch_id', 100)
record_id = validate_text(args.record_id, 'record_id', 100)
category = validate_text(args.proposed_category, 'proposed_category', 50)
reason = validate_text(args.reason, 'reason', 1000)
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
with connection:
connection.execute('BEGIN IMMEDIATE')
if connection.execute(
'SELECT 1 FROM allowed_category WHERE category = ?',
(category,),
).fetchone() is None:
raise AppError('unknown_category', '許可されていないカテゴリです')
if connection.execute(
'''
SELECT 1 FROM review_item
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone() is not None:
raise AppError('duplicate_item', 'レビュー項目がすでに存在します')
connection.execute(
'''
INSERT INTO review_item(
batch_id, record_id, proposed_category, reason, created_at
) VALUES (?, ?, ?, ?, ?)
''',
(batch_id, record_id, category, reason, int(time.time())),
)
return {'added': True, 'batch_id': batch_id, 'record_id': record_id}
def cmd_claim(args: argparse.Namespace) -> dict[str, Any]:
reviewer = validate_text(args.reviewer, 'reviewer', 100)
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
with connection:
connection.execute('BEGIN IMMEDIATE')
now = int(time.time())
row = connection.execute(
'''
SELECT batch_id, record_id, proposed_category, reason
FROM review_item
WHERE status = 'pending'
AND (claim_token IS NULL OR claim_expires_at <= ?)
ORDER BY created_at, batch_id, record_id
LIMIT 1
''',
(now,),
).fetchone()
if row is None:
result = {'claimed': False}
else:
token = secrets.token_urlsafe(24)
expires_at = now + args.lease_seconds
cursor = connection.execute(
'''
UPDATE review_item
SET claim_token = ?, claimed_by = ?,
claimed_at = ?, claim_expires_at = ?
WHERE batch_id = ? AND record_id = ?
AND status = 'pending'
AND (claim_token IS NULL OR claim_expires_at <= ?)
''',
(
token,
reviewer,
now,
expires_at,
row['batch_id'],
row['record_id'],
now,
),
)
if cursor.rowcount != 1:
raise AppError('claim_conflict', '取得状態が変わりました')
result = {
'claimed': True,
'batch_id': row['batch_id'],
'record_id': row['record_id'],
'proposed_category': row['proposed_category'],
'reason': row['reason'],
'claimed_by': reviewer,
'claimed_at': now,
'claim_expires_at': expires_at,
'claim_token': token,
}
return result
def cmd_renew(args: argparse.Namespace) -> dict[str, Any]:
token = load_claim_token(args)
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
with connection:
connection.execute('BEGIN IMMEDIATE')
now = int(time.time())
row = connection.execute(
'''
SELECT batch_id, record_id, claim_expires_at
FROM review_item
WHERE status = 'pending' AND claim_token = ?
''',
(token,),
).fetchone()
if row is None:
raise AppError('claim_not_found', '有効な取得情報がありません')
if row['claim_expires_at'] <= now:
raise AppError('claim_expired', 'ロックはすでに失効しています')
expires_at = now + args.lease_seconds
connection.execute(
'UPDATE review_item SET claim_expires_at = ? WHERE claim_token = ?',
(expires_at, token),
)
return {
'renewed': True,
'batch_id': row['batch_id'],
'record_id': row['record_id'],
'claim_expires_at': expires_at,
}
def cmd_release(args: argparse.Namespace) -> dict[str, Any]:
token = load_claim_token(args)
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
with connection:
connection.execute('BEGIN IMMEDIATE')
now = int(time.time())
row = connection.execute(
'''
SELECT batch_id, record_id, claim_expires_at
FROM review_item
WHERE status = 'pending' AND claim_token = ?
''',
(token,),
).fetchone()
if row is None:
raise AppError('claim_not_found', '有効な取得情報がありません')
if row['claim_expires_at'] <= now:
raise AppError('claim_expired', 'ロックはすでに失効しています')
connection.execute(
'''
UPDATE review_item
SET claim_token = NULL, claimed_by = NULL,
claimed_at = NULL, claim_expires_at = NULL
WHERE claim_token = ?
''',
(token,),
)
return {
'released': True,
'batch_id': row['batch_id'],
'record_id': row['record_id'],
}
def cmd_decide(args: argparse.Namespace) -> dict[str, Any]:
token = load_claim_token(args)
if args.decision == 'corrected':
if args.category is None:
raise AppError('category_required', 'correctedには--categoryが必要です')
category = validate_text(args.category, 'category', 50)
elif args.category is not None:
raise AppError(
'unexpected_category',
'acceptedとrejectedでは--categoryを指定しません',
)
else:
category = None
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
with connection:
connection.execute('BEGIN IMMEDIATE')
now = int(time.time())
row = connection.execute(
'''
SELECT batch_id, record_id, proposed_category,
claimed_by, claim_expires_at
FROM review_item
WHERE status = 'pending' AND claim_token = ?
''',
(token,),
).fetchone()
if row is None:
raise AppError('claim_not_found', '有効な取得情報がありません')
if row['claim_expires_at'] <= now:
raise AppError('claim_expired', 'ロックはすでに失効しています')
if args.decision == 'corrected':
if connection.execute(
'SELECT 1 FROM allowed_category WHERE category = ?',
(category,),
).fetchone() is None:
raise AppError('unknown_category', '許可されていないカテゴリです')
if category == row['proposed_category']:
raise AppError(
'category_unchanged',
'correctedでは提案と異なるカテゴリを指定してください',
)
final_category = category
elif args.decision == 'accepted':
final_category = row['proposed_category']
else:
final_category = None
cursor = connection.execute(
'''
UPDATE review_item
SET status = ?, final_category = ?, reviewed_by = ?,
decided_at = ?, claim_token = NULL, claimed_by = NULL,
claimed_at = NULL, claim_expires_at = NULL
WHERE batch_id = ? AND record_id = ?
AND status = 'pending' AND claim_token = ?
''',
(
args.decision,
final_category,
row['claimed_by'],
now,
row['batch_id'],
row['record_id'],
token,
),
)
if cursor.rowcount != 1:
raise AppError('decision_conflict', '確定状態が変わりました')
return {
'batch_id': row['batch_id'],
'record_id': row['record_id'],
'status': args.decision,
'final_category': final_category,
'reviewed_by': row['claimed_by'],
}
def cmd_show(args: argparse.Namespace) -> dict[str, Any]:
batch_id = validate_text(args.batch_id, 'batch_id', 100)
record_id = validate_text(args.record_id, 'record_id', 100)
with closing(connect_existing(args.db)) as connection:
verify_schema(connection)
row = connection.execute(
'''
SELECT batch_id, record_id, status, proposed_category,
final_category, reviewed_by, claim_token,
claimed_by, claim_expires_at
FROM review_item
WHERE batch_id = ? AND record_id = ?
''',
(batch_id, record_id),
).fetchone()
if row is None:
raise AppError('item_not_found', 'レビュー項目が見つかりません')
now = int(time.time())
if row['status'] != 'pending':
claim_state = 'not_applicable'
elif row['claim_token'] is None:
claim_state = 'unclaimed'
elif row['claim_expires_at'] <= now:
claim_state = 'expired'
else:
claim_state = 'active'
return {
'batch_id': row['batch_id'],
'record_id': row['record_id'],
'status': row['status'],
'proposed_category': row['proposed_category'],
'final_category': row['final_category'],
'reviewed_by': row['reviewed_by'],
'claimed_by': row['claimed_by'],
'claim_state': claim_state,
}
def add_claim_source(parser: argparse.ArgumentParser) -> None:
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument('--token')
group.add_argument('--claim-json', type=Path)
def build_parser() -> argparse.ArgumentParser:
parser = JsonArgumentParser(description='期限付きレビューキュー')
subparsers = parser.add_subparsers(dest='command', required=True)
init_parser = subparsers.add_parser('init')
init_parser.add_argument('db', type=Path)
init_parser.add_argument('--category', action='append', required=True)
init_parser.set_defaults(handler=cmd_init)
add_parser = subparsers.add_parser('add')
add_parser.add_argument('db', type=Path)
add_parser.add_argument('--batch-id', required=True)
add_parser.add_argument('--record-id', required=True)
add_parser.add_argument('--proposed-category', required=True)
add_parser.add_argument('--reason', required=True)
add_parser.set_defaults(handler=cmd_add)
claim_parser = subparsers.add_parser('claim')
claim_parser.add_argument('db', type=Path)
claim_parser.add_argument('--reviewer', required=True)
claim_parser.add_argument(
'--lease-seconds', type=lease_seconds, default=300
)
claim_parser.set_defaults(handler=cmd_claim)
renew_parser = subparsers.add_parser('renew')
renew_parser.add_argument('db', type=Path)
renew_parser.add_argument(
'--lease-seconds', type=lease_seconds, default=300
)
add_claim_source(renew_parser)
renew_parser.set_defaults(handler=cmd_renew)
release_parser = subparsers.add_parser('release')
release_parser.add_argument('db', type=Path)
add_claim_source(release_parser)
release_parser.set_defaults(handler=cmd_release)
decide_parser = subparsers.add_parser('decide')
decide_parser.add_argument('db', type=Path)
decide_parser.add_argument(
'--decision',
choices=['accepted', 'corrected', 'rejected'],
required=True,
)
decide_parser.add_argument('--category')
add_claim_source(decide_parser)
decide_parser.set_defaults(handler=cmd_decide)
show_parser = subparsers.add_parser('show')
show_parser.add_argument('db', type=Path)
show_parser.add_argument('--batch-id', required=True)
show_parser.add_argument('--record-id', required=True)
show_parser.set_defaults(handler=cmd_show)
return parser
def main() -> int:
try:
args = build_parser().parse_args()
result = args.handler(args)
except AppError as exc:
emit(
{'ok': False, 'error': exc.code, 'message': exc.message},
error=True,
)
return exc.exit_code
except sqlite3.Error as exc:
emit(
{
'ok': False,
'error': 'database_error',
'message': f'SQLite処理に失敗しました: {exc}',
},
error=True,
)
return 3
except OSError as exc:
emit(
{
'ok': False,
'error': 'io_error',
'message': f'ファイル操作に失敗しました: {exc}',
},
error=True,
)
return 3
except KeyboardInterrupt:
emit(
{'ok': False, 'error': 'interrupted', 'message': '中断しました'},
error=True,
)
return 130
emit({'ok': True, **result})
return 0
if __name__ == '__main__':
raise SystemExit(main())
1. データベースと確認用データを作る
既存の第163回用データベースではなく、新しいファイルを指定します。
python3 review_leases.py init review-leases.db --category 配送 --category 請求 --category その他
python3 review_leases.py add review-leases.db --batch-id batch-20261005 --record-id C-001 --proposed-category 配送 --reason '提案カテゴリを元データと照合する'
成功時の出力は次のとおりです。
{"ok":true,"database":"review-leases.db","category_count":3}
{"ok":true,"added":true,"batch_id":"batch-20261005","record_id":"C-001"}
initは既存ファイルを拒否します。カテゴリの追加や変更を行う管理コマンドは、この最小実装には含めていません。
2. レビュー項目を取得する
取得結果にはレビュー理由とロックトークンが含まれます。次の例では、未使用のファイルへ保存します。シェルのリダイレクトは既存ファイルを上書きすることがあるため、既存ファイルを出力先に指定しないでください。
python3 review_leases.py claim review-leases.db --reviewer tanaka --lease-seconds 300 > claim-tanaka.json
python3 -m json.tool claim-tanaka.json
取得に成功した場合、claim-tanaka.jsonには対象ID、提案カテゴリ、理由、担当者、取得時刻、有効期限、トークンが入ります。時刻とトークンは実行ごとに変わります。最初のコマンドの終了コードが0であり、JSONのclaimedがtrueであることを確認してから確定操作へ進みます。
この例では未確定項目が1件だけなので、有効期限内に別の担当者が取得すると、次の結果になります。
python3 review_leases.py claim review-leases.db --reviewer suzuki --lease-seconds 300
{"ok":true,"claimed":false}
claimed=falseは異常終了ではなく、現在取得できる項目がないことを表します。この結果にはトークンがないため、decideには使えません。
3. 必要なら期限を更新または解放する
レビューを続ける場合は、有効期限が切れる前にrenewを実行します。
python3 review_leases.py renew review-leases.db --claim-json claim-tanaka.json --lease-seconds 300
新しい期限は、以前の期限ではなくrenewの実行時刻を基準に計算されます。指定した秒数が現在の残り時間より短い場合は期限が前倒しされるため、通常は取得時と同じ値を指定し、失効間際になる前に更新します。更新結果のJSONに新しいclaim_expires_atが返りますが、保存済みのclaim-tanaka.json内の時刻は書き換わりません。トークンはそのまま利用できます。
確定せず別の担当者へ戻す場合はreleaseを使います。次のコマンドは、手順4とは別の分岐例です。手順4で同じ取得結果を使って確定する場合は実行しません。
python3 review_leases.py release review-leases.db --claim-json claim-tanaka.json
失効後のトークンでは更新も解放もできません。失効した項目は、次のclaimによって新しい担当者とトークンへ置き換えられます。
4. レビュー結果を確定する
提案された配送を請求へ修正する場合は、許可済みカテゴリを指定します。
python3 review_leases.py decide review-leases.db --claim-json claim-tanaka.json --decision corrected --category 請求
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","final_category":"請求","reviewed_by":"tanaka"}
判定ごとの引数は次のとおりです。
accepted：提案カテゴリを採用する。--categoryは指定しないcorrected：提案と異なる許可カテゴリを--categoryで指定するrejected：最終カテゴリを持たせない。--categoryは指定しない
確定後はトークンがデータベースから消えるため、同じ取得結果を再利用したdecideはclaim_not_foundで失敗します。失効後、まだ再取得されていないトークンを使った場合はclaim_expiredになります。
5. 保存状態を確認する
python3 review_leases.py show review-leases.db --batch-id batch-20261005 --record-id C-001
{"ok":true,"batch_id":"batch-20261005","record_id":"C-001","status":"corrected","proposed_category":"配送","final_category":"請求","reviewed_by":"tanaka","claimed_by":null,"claim_state":"not_applicable"}
showはトークンを表示しません。未確定項目のclaim_stateは、unclaimed、active、expiredのいずれかです。
同時取得を防ぐ仕組み
claimは、対象の検索前にBEGIN IMMEDIATEを実行します。SQLiteの書き込みトランザクションを確保してから、取得可能な行の検索と更新を同じトランザクション内で行います。このCLIだけがデータベースを更新する構成では、別プロセスが同じ行を同時に取得することを防げます。
ロック待機時間は5秒です。競合が長引いた場合はSQLiteエラーとなり、終了コード3を返すことがあります。高い並列処理性能を目的とした設計ではありません。
| 終了コード | 意味 |
|---|---|
| 0 | 操作成功。取得対象がないclaimed=falseも含む |
| 2 | 引数、カテゴリ、トークン、取得結果ファイル、有効期限、状態遷移など、入力または業務条件による拒否 |
| 3 | SQLite処理、または想定外のファイル操作の失敗 |
| 130 | Ctrl+Cによる中断 |
実運用へ入れる前の制約
- 取得結果とデータベースにはトークンが含まれるため、アクセスできる利用者をOSの権限で限定する
- 端末の時刻変更は有効期限の判定へ影響するため、各プロセスが同じ時刻基準を使える環境で運用する
- プロセスが異常終了した場合、明示的な解放は行われず、有効期限までロックが残る
- 期限切れ後に元担当者の作業画面を停止する仕組みはない。別の担当者が再取得した場合、元担当者からの確定は拒否されるが、レビュー作業自体は重複する可能性がある
- ネットワーク共有上のSQLiteは、ファイルロックの挙動を検証できない限り使用しない
- 多数の同時利用者、利用者認証、権限分離、通知が必要な場合は、SQLiteだけで要件を満たそうとしない
- この実装が保存するのは現在状態だけであり、変更不能な監査ログは作らない。監査要件がある場合は、決定履歴を別の仕組みで記録する
まずはサンプル1件で、取得中の再取得拒否、期限内の確定、解放後の再取得を確認します。既存のレビューキューへ統合する場合は、データベースを退避したうえで、実際のテーブル名と制約に合わせた移行処理を別に用意してください。