simple-db-hds 1.0.0

A lightweight, cross-platform, multi-process safe JSON database

SimpleDB 使用文档

一个轻量级、跨平台、支持多进程并发的 JSON 数据库，基于 TinyDB 实现。

特性

• 跨平台: 支持 Windows、Linux、macOS
• 多进程安全: 文件锁机制保证并发读写
• 简单易用: 无需 SQL，通过 Python 表达式查询
• 零依赖: 仅依赖 TinyDB 和标准库

安装

pip install tinydb

快速开始

from simple_db import SimpleDB

# 初始化数据库
db = SimpleDB("mydb.json")

# 插入数据
db.insert({"name": "张三", "age": 25})
db.insert({"name": "李四", "age": 30})

# 查询
results = db.query(db.Q.age > 28)
print(results)  # [{'name': '李四', 'age': 30}]

db.close()

---

API 参考

初始化

db = SimpleDB(db_file="data.json", timeout=30)

参数	类型	默认值	说明
db_file	str	"data.json"	数据库文件路径
timeout	float	30	文件锁超时时间(秒)

---

插入数据

insert(item: dict) -> int

插入单条记录，返回记录ID。

db_id = db.insert({"name": "张三", "age": 25})

insert_many(items: list[dict]) -> list[int]

批量插入多条记录，返回记录ID列表。

ids = db.insert_many([
    {"name": "张三", "age": 25},
    {"name": "李四", "age": 30},
])

---

查询数据

all() -> list[dict]

获取所有记录。

all_records = db.all()

query(condition) -> list[dict]

条件查询，支持链式调用。

# 基础比较
db.query(db.Q.age > 18)           # age > 18
db.query(db.Q.name == "张三")      # name == "张三"
db.query(db.Q.age >= 21)          # age >= 21
db.query(db.Q.age < 30)           # age < 30
db.query(db.Q.age <= 25)          # age <= 25
db.query(db.Q.name != "张三")     # name != "张三"

# 逻辑组合
db.query((db.Q.age > 20) & (db.Q.age < 30))   # AND
db.query((db.Q.city == "北京") | (db.Q.city == "上海"))  # OR
db.query(~db.Q.active)                          # NOT

fuzzy_query(field: str, keyword: str) -> list[dict]

模糊查询（包含匹配）。

# 查找 name 包含 "张" 的记录
results = db.fuzzy_query("name", "张")

# 查找 skills 包含 "Python" 的记录
results = db.fuzzy_query("skills", "Python")

page(page: int = 1, page_size: int = 10) -> list[dict]

分页查询。

# 第1页，每页10条
page1 = db.page(page=1, page_size=10)

# 第2页，每页5条
page2 = db.page(page=2, page_size=5)

---

更新数据

update(condition, new_data: dict) -> list[int]

更新符合条件的所有记录。

# 更新所有 age > 25 的记录
db.update(db.Q.age > 25, {"age": 26})

# 更新单个字段
db.update(db.Q.name == "张三", {"city": "成都"})

# 更新多个字段
db.update(db.Q.name == "李四", {"age": 31, "city": "深圳"})

---

删除数据

delete(condition) -> list[int]

删除符合条件的所有记录，返回被删除的记录ID。

# 删除单个记录
db.delete(db.Q.name == "张三")

# 批量删除
db.delete(db.Q.age < 18)

# 删除所有（慎用）
db.delete(db.Q.name.test(lambda x: True))

---

其他操作

clear()

清空当前表的所有数据。

db.clear()

close()

关闭数据库连接（建议操作完成后调用）。

db.close()

---

多表支持

# 访问不同表
users = db.db.table("users")
users.insert({"name": "张三", "role": "admin"})

settings = db.db.table("settings")
settings.insert({"theme": "dark"})

---

多进程并发

多个 Python 进程可以同时读写同一个数据库文件，文件锁保证数据一致性。

# process_a.py
from simple_db import SimpleDB
db = SimpleDB("shared.json")
db.insert({"from": "进程A"})
db.close()

# process_b.py
from simple_db import SimpleDB
db = SimpleDB("shared.json")
db.insert({"from": "进程B"})
db.close()

python process_a.py &
python process_b.py &

---

完整示例

from simple_db import SimpleDB

db = SimpleDB("demo.json")

# 初始化数据
db.insert_many([
    {"name": "张三", "age": 25, "city": "北京", "skills": ["Python", "Go"]},
    {"name": "李四", "age": 30, "city": "上海", "skills": ["Java", "Rust"]},
    {"name": "王五", "age": 28, "city": "北京", "skills": ["Python", "JavaScript"]},
])

Q = db.Q

# 查询北京用户
beijingers = db.query(Q.city == "北京")
print(f"北京用户: {[u['name'] for u in beijingers]}")

# 查询会Python的人
pythonistas = db.fuzzy_query("skills", "Python")
print(f"会Python: {[u['name'] for u in pythonistas]}")

# 更新年龄
db.update(Q.name == "张三", {"age": 26})

# 删除记录
db.delete(Q.name == "李四")

# 分页展示
for i, page in enumerate(db.page(page_size=2), 1):
    print(f"第{i}页: {page}")

db.close()

---

注意事项

1. 文件锁超时: 默认30秒，超时后抛出 TimeoutError
2. JSON格式: 数据以缩进2空格格式存储
3. 锁文件: 自动创建 .lockfile 后缀的锁文件
4. 性能: 适合小数据量，大数据建议使用 SQLite 或 PostgreSQL