88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""JSON 存储引擎：一个文件 = 一张表。
|
||
|
||
当前阶段用本地 JSON 文件代替数据库。每个文件是一个 JSON 数组，
|
||
元素为一行记录；存储层只做最朴素的"读全量/写全量 + 原子替换"，
|
||
业务逻辑全部收敛在 ``repositories.py``。
|
||
|
||
未来数据库映射（下一阶段据此设计）：
|
||
JsonTable(path) -> 一张数据库表（表名 = 文件名）
|
||
数组元素 -> 一行记录（主键 = 记录里的 ``id`` 字段）
|
||
记录字段 -> 一列
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import threading
|
||
from pathlib import Path
|
||
from typing import Callable
|
||
|
||
|
||
class JsonTable:
|
||
"""对单个 JSON 数组文件的最小 CRUD 封装。
|
||
|
||
- 写入使用「写临时文件 + rename 替换」，避免崩溃/并发读到半截数据。
|
||
- 内部用 RLock 保证同一进程内的读写互斥（演示级并发足够）。
|
||
"""
|
||
|
||
def __init__(self, path: Path):
|
||
self.path = path
|
||
self._lock = threading.RLock()
|
||
|
||
def all(self) -> list[dict]:
|
||
"""读全部记录。文件不存在/损坏时返回空表（下次写入时重建）。"""
|
||
with self._lock:
|
||
if not self.path.exists():
|
||
return []
|
||
try:
|
||
with open(self.path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
except (json.JSONDecodeError, OSError):
|
||
return []
|
||
return data if isinstance(data, list) else []
|
||
|
||
def save(self, records: list[dict]) -> None:
|
||
"""整体写回。原子：先写 ``<name>.tmp`` 再 rename。"""
|
||
with self._lock:
|
||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = self.path.with_name(self.path.name + ".tmp")
|
||
with open(tmp, "w", encoding="utf-8") as fh:
|
||
json.dump(records, fh, ensure_ascii=False, indent=2)
|
||
tmp.replace(self.path)
|
||
|
||
# ------------------------------------------------------------------
|
||
# 便捷操作
|
||
# ------------------------------------------------------------------
|
||
def find(self, predicate: Callable[[dict], bool]) -> dict | None:
|
||
"""返回第一条满足条件的记录，没有则 None。"""
|
||
for record in self.all():
|
||
if predicate(record):
|
||
return record
|
||
return None
|
||
|
||
def insert(self, record: dict) -> dict:
|
||
"""追加一条记录并返回它。"""
|
||
records = self.all()
|
||
records.append(record)
|
||
self.save(records)
|
||
return record
|
||
|
||
def update(self, record_id: str, fields: dict) -> dict | None:
|
||
"""按 ``id`` 合并更新字段，返回更新后的记录（不存在则 None）。"""
|
||
records = self.all()
|
||
for record in records:
|
||
if record.get("id") == record_id:
|
||
record.update(fields)
|
||
self.save(records)
|
||
return record
|
||
return None
|
||
|
||
def delete_matching(self, predicate: Callable[[dict], bool]) -> int:
|
||
"""删除所有满足条件的记录，返回删除条数。"""
|
||
records = self.all()
|
||
kept = [r for r in records if not predicate(r)]
|
||
removed = len(records) - len(kept)
|
||
if removed:
|
||
self.save(kept)
|
||
return removed
|