database - PythonIDE Docs

MiniApp 作用域内的原生 SQLite 持久化：可查询的历史、收藏、缓存与业务表。

边界：数据库文件自动放在当前 MiniApp 数据目录，并由宿主原生 SQLite 桥管理连接、statement 生命周期和事务。小型开关/主题用 storage；token/密码用 keychain。只传文件名（如 "media"），不要传绝对路径或 ../；MiniApp 不要直接 import sqlite3。

#模块概览

项	说明
导入	`import database`
适合做什么	播放历史、收藏、离线缓存、可排序列表、业务表
两条路径	大多数场景用 `collection()`；需要索引/JOIN 用 `open()` + SQL
调用时机	启动时读取、按钮回调写入；长任务结束可 `close()`
安全写入	SQL 用 `?` 参数绑定，不要拼接用户输入

#快速开始

下面脚本用 Collection API 写入并列出收藏：

python

import database

favorites = database.collection("favorites")

favorites.upsert("movie-1", {"title": "示例电影", "rating": 9.2})
print(favorites.get("movie-1"))
print(favorites.count())

for item in favorites.list(order_by="updated_at desc", limit=10):
    print(item["title"], item["rating"])

#AppUI 示例

用 Collection 管理列表示例数据；按钮触发增删查。

python

import appui
import database

favorites = database.collection("demo_favorites")

state = appui.State(
    status="等待操作",
    count="0",
    items=[],
)


def item_key(row):
    return row["id"]


def item_row(row):
    return appui.HStack([
        appui.Text(row["title"]),
        appui.Spacer(),
        appui.Text(row["rating"]).foreground_color("secondaryLabel"),
    ])


def refresh_list():
    rows = []
    for entry in favorites.items(order_by="updated_at desc", limit=10):
        value = entry.get("value") or {}
        rows.append({
            "id": entry["key"],
            "title": value.get("title", entry["key"]),
            "rating": f"⭐ {value.get('rating', '—')}",
        })
    state.batch_update(
        status=f"共 {favorites.count()} 条记录",
        count=str(favorites.count()),
        items=rows,
    )


def add_sample():
    key = f"item-{favorites.count() + 1}"
    favorites.upsert(key, {"title": f"示例 {key}", "rating": 8.5})
    refresh_list()


def clear_all():
    favorites.clear()
    refresh_list()
    state.status = "已清空 collection"


def body():
    return appui.NavigationStack(
        appui.List([
            appui.Section("操作", [
                appui.Button("添加示例", action=add_sample)
                .button_style("bordered_prominent"),
                appui.Button("刷新列表", action=refresh_list),
                appui.Button("清空", action=clear_all, role="destructive"),
                appui.LabeledContent("状态", value=state.status),
            ]),
            appui.Section("列表", [
                appui.ForEach(state.items, row_builder=item_row, key=item_key),
            ], footer=f"当前 {state.count} 条 · collection: demo_favorites"),
        ]).navigation_title("数据库")
    )


appui.run(body, state=state)

#API 参考

#速查

API	作用
`collection(name)`	默认库中的 JSON 文档集合（推荐）
`open(name)` / `connect(name)`	打开 MiniApp 作用域 SQLite
`database_path(name)`	调试用的数据库绝对路径
`close_default_database()` / `close_all()`	关闭默认库或当前进程打开的库
`Collection.upsert/get/delete`	增删改查 JSON 记录
`Collection.items/list`	排序分页列表
`Database.execute/query/transaction`	原生 SQL 操作

#选型

目标	首选 API
收藏、历史、缓存、源列表	`database.collection(name)`
索引、JOIN、迁移、事务	`database.open(name)` + SQL
少量开关、主题	storage
token、密码	keychain

#Collection API

大多数 MiniApp 的首选，底层 SQLite 存 JSON 文档：

python

import database

history = database.collection("history")
history.upsert("ep-42", {"id": "ep-42", "title": "第 42 集", "position": 128.5})
row = history.get("ep-42")
items = history.list(order_by="updated_at desc", limit=20)
history.delete("ep-42")
history.clear()
n = history.count()

API	说明
`upsert(key, value)`	插入或更新 JSON
`get(key, default=None)`	读取；不存在返回 `default`
`delete(key)`	删除一条
`clear()`	清空当前 collection
`count()`	记录数
`items(order_by, limit, offset)`	含 `key`/`value`/时间戳
`list(order_by, limit, offset)`	只返回 value 列表
`migrate_from_storage(key, key_field="id")`	从 storage JSON 迁移

order_by 支持：updated_at desc/asc、created_at desc/asc、key desc/asc。

#SQL API

需要明确表结构时使用：

python

import database

db = database.open("media")
db.executescript("""
CREATE TABLE IF NOT EXISTS episodes (
    id TEXT PRIMARY KEY,
    title TEXT NOT NULL,
    position REAL NOT NULL DEFAULT 0
);
""")

with db.transaction():
    db.execute(
        "INSERT OR REPLACE INTO episodes(id, title, position) VALUES (?, ?, ?)",
        ["ep-42", "第 42 集", 128.5],
    )

row = db.query_one("SELECT * FROM episodes WHERE id = ?", ["ep-42"])
db.close()

API	说明
`execute(sql, params)`	执行 SQL，返回影响行数
`query(sql, params)`	返回 `list[dict]`
`query_one(sql, params, default)`	单行或默认值
`scalar(sql, params, default)`	第一列标量
`transaction()`	原子事务，异常回滚
`user_version`	schema 版本号（`PRAGMA user_version`）
`close()`	关闭当前连接

database.open("media") 自动变为 media.db，拒绝 ../ 和绝对路径。

#从 storage 迁移

python

history = database.collection("history")
count = history.migrate_from_storage("video.history", key_field="id", remove=True)
print("已迁移", count, "条")

#常见错误

错误写法	后果	修正
大列表塞进 `storage`	慢、难查询	用 `collection()`
token 写进 SQLite	不安全	用 keychain
`open("../data.db")`	路径被拒绝	只传 `"media"` 等文件名
直接 `import sqlite3`	真机可能崩溃或绕过宿主隔离	用 `import database`
SQL 拼接用户输入	注入风险	用 `?` 参数绑定
长时间不 `close()`	连接泄漏	任务结束时关闭

#相关文档

文档	用途
storage	小型键值偏好
keychain	敏感凭据
原生能力入口	MiniApp 场景配方

#预期效果

运行示例后，界面应出现文档描述的目标结果；若与预期不符，先看「失败路径」并按返回值或日志排查。