zzhyyyyy
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""SQLite 持久化层。占坑期方案,够用即可。
|
|
|
|
- 单文件 sqlite3,默认 `/opt/duobibi-server/data.db`(env `DUOBIBI_DB_PATH` 可覆盖)
|
|
- 用 stdlib sqlite3,不引入 SQLAlchemy(占坑期不需要 ORM)
|
|
- 表结构:
|
|
- `user`:phone 当主键(占坑期账号体系)
|
|
- `wish_item`:用户的心愿单,phone 关联 user,deleted_at 软删,updated_at 用于同步冲突解决
|
|
|
|
并发说明:sqlite3 默认每个连接独占,FastAPI 是 thread pool 跑 sync handler。
|
|
每次请求新开 connection + use as ctx manager,简单可靠。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sqlite3
|
|
from contextlib import contextmanager
|
|
from typing import Iterator
|
|
|
|
DB_PATH = os.environ.get("DUOBIBI_DB_PATH", "/opt/duobibi-server/data.db")
|
|
|
|
|
|
def _connect() -> sqlite3.Connection:
|
|
conn = sqlite3.connect(DB_PATH, timeout=10.0, isolation_level=None)
|
|
conn.row_factory = sqlite3.Row
|
|
# WAL 提升并发读写;FastAPI thread pool 下多线程访问同一文件时有用
|
|
conn.execute("PRAGMA journal_mode=WAL")
|
|
conn.execute("PRAGMA foreign_keys=ON")
|
|
return conn
|
|
|
|
|
|
@contextmanager
|
|
def get_conn() -> Iterator[sqlite3.Connection]:
|
|
"""每次请求开一个新 connection,with 块结束自动 close。"""
|
|
conn = _connect()
|
|
try:
|
|
yield conn
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def init_schema() -> None:
|
|
"""启动时调用,IF NOT EXISTS 建表。改 schema 需手动 migration(写额外 SQL)。"""
|
|
with get_conn() as conn:
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS user (
|
|
phone TEXT PRIMARY KEY,
|
|
created_at INTEGER NOT NULL,
|
|
last_seen_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS wish_item (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
phone TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
target_price REAL,
|
|
note TEXT,
|
|
notify_enabled INTEGER NOT NULL DEFAULT 1,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
deleted_at INTEGER,
|
|
FOREIGN KEY (phone) REFERENCES user(phone) ON DELETE CASCADE
|
|
);
|
|
|
|
-- 按用户拉取 / 增量同步主索引
|
|
CREATE INDEX IF NOT EXISTS idx_wish_phone_updated
|
|
ON wish_item(phone, updated_at DESC);
|
|
"""
|
|
)
|
|
|
|
|
|
def upsert_user(phone: str, now_ms: int) -> None:
|
|
"""登录成功时调:有则更新 last_seen_at,无则插入。"""
|
|
with get_conn() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO user (phone, created_at, last_seen_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(phone) DO UPDATE SET last_seen_at = excluded.last_seen_at
|
|
""",
|
|
(phone, now_ms, now_ms),
|
|
)
|