81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""
|
||
database.py
|
||
数据库模块：负责 SQLite 连接、表初始化和基础 CRUD 操作。
|
||
使用 Python 内置 sqlite3，无需额外 ORM，保持轻量。
|
||
"""
|
||
|
||
import sqlite3
|
||
import os
|
||
from contextlib import contextmanager
|
||
from datetime import datetime
|
||
|
||
# 数据库文件路径（放在当前目录下，轻量本地运行）
|
||
DB_PATH = os.path.join(os.path.dirname(__file__), "device_platform.db")
|
||
|
||
|
||
def get_db_connection() -> sqlite3.Connection:
|
||
"""
|
||
创建并返回一个 SQLite 连接对象。
|
||
设置 row_factory 为 sqlite3.Row，使查询结果可以通过列名访问。
|
||
"""
|
||
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
|
||
@contextmanager
|
||
def get_db():
|
||
"""
|
||
上下文管理器，用于在 API 接口中安全获取和释放数据库连接。
|
||
用法：
|
||
with get_db() as db:
|
||
db.execute(...)
|
||
"""
|
||
conn = get_db_connection()
|
||
try:
|
||
yield conn
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def init_db():
|
||
"""
|
||
初始化数据库：创建设备表（如果不存在）。
|
||
设备表字段：
|
||
- sn: 设备序列号，主键，唯一标识一台设备
|
||
- status: 设备状态（待激活 / 已激活 / 已禁用）
|
||
- activated_at: 激活时间，为空表示尚未激活
|
||
- created_at: 记录创建时间
|
||
"""
|
||
with get_db() as db:
|
||
db.execute("""
|
||
CREATE TABLE IF NOT EXISTS devices (
|
||
sn TEXT PRIMARY KEY NOT NULL,
|
||
status TEXT NOT NULL DEFAULT '待激活',
|
||
activated_at TEXT,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime'))
|
||
)
|
||
""")
|
||
db.commit()
|
||
print(f"[DB] 数据库已初始化: {DB_PATH}")
|
||
|
||
|
||
def seed_demo_data():
|
||
"""
|
||
插入演示数据，方便新手直接体验接口。
|
||
如果已存在相同 SN，则忽略（INSERT OR IGNORE）。
|
||
"""
|
||
demo_devices = [
|
||
("GD30-20260507-001", "待激活"),
|
||
("GD30-20260507-002", "待激活"),
|
||
("MT-20260507-003", "已禁用"),
|
||
]
|
||
with get_db() as db:
|
||
for sn, status in demo_devices:
|
||
db.execute(
|
||
"INSERT OR IGNORE INTO devices (sn, status) VALUES (?, ?)",
|
||
(sn, status)
|
||
)
|
||
db.commit()
|
||
print("[DB] 演示数据已插入")
|