Taipy Core与SQLite数据节点连接失败，求可行示例

前置准备 #

• 安装依赖：pip install taipy-core
• SQLite3 是 Python 标准库，无需额外安装

完整可运行代码 #

import taipy as tp
from taipy.config import Config
import sqlite3

# 初始化SQLite数据库并创建目标表（首次运行必须）
def init_db():
    conn = sqlite3.connect("taipy.db")
    cursor = conn.cursor()
    # 创建sales表
    cursor.execute('''CREATE TABLE IF NOT EXISTS sales
                      (id INTEGER PRIMARY KEY, amount INTEGER)''')
    # 插入测试数据（避免重复插入）
    cursor.execute("INSERT OR IGNORE INTO sales (id, amount) VALUES (1, 100), (2, 200)")
    conn.commit()
    conn.close()

# 定义写入查询构建器（必须实现，用于数据写入逻辑）
def write_query_builder(data, data_node):
    return f"INSERT INTO sales (id, amount) VALUES (?, ?) ON CONFLICT(id) DO UPDATE SET amount=?"

# 任务处理函数
def double(input_data):
    # 输入是从SQL读取的DataFrame，处理后返回
    input_data['amount'] = input_data['amount'] * 2
    return input_data

# 初始化数据库
init_db()

# 配置输入SQL数据节点
sales_history_cfg = Config.configure_sql_data_node(
    id="sales_history",
    db_username="",
    db_password="",
    db_name="taipy.db",  # SQLite必须指定带.db后缀的文件路径
    db_engine="sqlite",
    read_query="SELECT * FROM sales",
    write_query_builder=write_query_builder
)

# 配置输出SQL数据节点（用于写入处理后的数据）
output_data_node_cfg = Config.configure_sql_data_node(
    id="doubled_sales",
    db_username="",
    db_password="",
    db_name="taipy.db",
    db_engine="sqlite",
    read_query="SELECT * FROM sales",
    write_query_builder=write_query_builder
)

# 配置任务
task_cfg = Config.configure_task(
    id="double_task",
    function=double,
    input=[sales_history_cfg],
    output=[output_data_node_cfg]
)

# 配置管道和场景
pipeline_cfg = Config.configure_pipeline(id="my_pipeline", task_configs=[task_cfg])
scenario_cfg = Config.configure_scenario(id="my_scenario", pipeline_configs=[pipeline_cfg])

# 运行Taipy Core
if __name__ == "__main__":
    tp.Core().run()

关键配置说明 #

• db_name 设置：SQLite 的 db_name 必须指定带 .db 后缀的文件路径，仅写 taipy 会导致 Taipy 无法识别为 SQLite 数据库文件，引发连接失败。
• write_query_builder 实现：必须提供合法的写入逻辑，示例中使用 INSERT OR IGNORE 避免重复插入，同时支持更新现有数据。
• 数据库初始化：首次运行需要创建目标表并插入测试数据，否则读取查询会因表不存在报错。

运行验证 #

1. 运行代码后，当前目录会生成 taipy.db 文件。
2. 用 SQLite 工具打开该文件，可查看 sales 表数据是否被成功读取并更新（amount 字段变为原数值的 2 倍）。

内容的提问来源于stack exchange，提问作者Markus