Taipy Core 深度解析数据集成、场景管理与版本管理后端引擎【免费下载链接】taipyTurns Data and AI algorithms into production-ready web applications in no time.项目地址: https://gitcode.com/GitHub_Trending/ta/taipy导读Taipy Core 是 Taipy 生态中的核心后端引擎负责将数据科学家手中的算法与数据转化为可生产化的业务应用后端。本文以仓库中 taipy/core/package_desc.md 为骨架结合 taipy-core 4.2.0 的源码与测试系统讲解其四大核心能力数据集成与管理、场景管理、What-if 分析、版本管理、实体模型、公共 API、Orchestrator 服务与安装测试方式帮助你快速理解并上手 taipy-core。一、Taipy Core 是什么Taipy 是一个用于创建业务应用Business Applications的 Python 库。而 taipy-core 是其中面向业务的功能包其定位可以从包描述与源码__init__.py中得到确认The Taipycorepackage provides powerful, customized,>pip install taipy-core当前仓库中 taipy/core/version.json 记录的版本为4.2.0可作为安装时版本判断的参考。2.2 安装开发版通过 pip 与 git 从 taipy 仓库安装开发版pip install githttps://gitgithub.com/Avaiga/taipy该命令会在当前 Python 环境中安装 taipy 包的开发版本及其全部依赖包含 taipy-core 包。你也可以通过克隆仓库来查看源码或参与改进git clone https://github.com/Avaiga/taipy.git克隆后taipy-core 的源码位于仓库的taipy/core目录在当前工作副本中即 taipy/core/。2.3 可选依赖从 taipy/core/setup.py 可以看到taipy-core 为不同的数据后端提供了extras_require可选依赖按需安装可以启用对应数据节点类型的支持extras_require { mssql: [pyodbc4,4.1], mysql: [pymysql1,1.1], postgresql: [psycopg22.9,2.10], parquet: [fastparquet2022.11.0, pyarrow16.0.0,19.0], s3: [boto31.29.1], mongo: [pymongo[srv]4.2.0,5.0], }例如安装 PostgreSQL 支持pip install taipy-core[postgresql]三、核心实体模型从 Data Node 到 Scenariotaipy-core 的所有能力都建立在几个核心实体之上每个实体在源码中都有独立模块Data Node数据节点对底层数据文件、数据库、内存对象等的统一抽象负责数据的读写与编辑历史。实现见 taipy/core/data/data_node.py类型覆盖 CSV、Excel、Parquet、Pickle、JSON、SQL、SQL Table、MongoDB、AWS S3、Generic 与 In-memory见 taipy/core/data/。Task任务一个可执行的 Python 函数及其输入输出 Data Node 的绑定是编排的最小执行单元。实现见 taipy/core/task/task.py。Sequence序列一组按序执行的任务集合。实现见 taipy/core/sequence/sequence.py。Scenario场景业务分析的顶层实体包含若干 Sequence 与 Data Node代表一次完整的业务运行假设例如夏季促销方案。实现见 taipy/core/scenario/scenario.py。Cycle周期当场景配置了频率如每日、每周时按创建日期归入对应的周期便于周期性的业务复盘。实现见 taipy/core/cycle/cycle.py。Job作业Task 被提交后生成的执行记录携带状态机见 taipy/core/job/status.py与执行结果。Submission提交一次提交操作的统一记录包含所创建的全部 Job。实现见 taipy/core/submission/submission.py。这些实体统一由各 Manager 管理如 taipy/core/scenario/_scenario_manager.py、taipy/core/data/_data_manager.py并通过 Repository 层持久化见 taipy/core/_repository/默认采用文件系统存储。四、第一步用 Config 设计应用package_desc.md对应的__init__.py明确指出使用这些功能的第一步是通过Config单例类设计应用的特性与行为To use such functionalities, the first step consists of setting up the Taipy configuration to design your applications characteristics and behaviors. Use theConfig^singleton class (fromtaipy.common.config) to configure your application.Config单例来自taipy.common.config其在仓库中的类型声明见 taipy/common/config/config.pyi常用的配置入口包括Config.configure_data_node(...)配置数据节点存储类型、默认路径等声明见 config.pyi对应配置类 taipy/core/config/data_node_config.py。Config.configure_task(...)将函数与输入/输出数据节点绑定成任务声明见 config.pyi对应配置类 taipy/core/config/task_config.py。Config.configure_scenario(...)组合任务序列并可选指定周期频率声明见 config.pyi对应配置类 taipy/core/config/scenario_config.py。Config.configure_job_executions(...)与Config.configure_core(...)分别控制作业执行模式与 Core 全局行为如开发模式、版本管理策略声明见 config.pyi 与 config.pyi。一个最小配置与使用骨架如下import taipy as tp from taipy.common.config import Config # 1. 配置数据节点 input_cfg Config.configure_data_node(input, storage_typecsv, default_pathdata.csv) output_cfg Config.configure_data_node(output, storage_typecsv, default_pathresult.csv) # 2. 将业务函数包装为任务 def process_data(input_data): return input_data * 2 task_cfg Config.configure_task(process, process_data, inputinput_cfg, outputoutput_cfg) # 3. 配置场景 scenario_cfg Config.configure_scenario(my_scenario, task_configs[task_cfg]) # 4. 创建并运行场景 scenario tp.create_scenario(scenario_cfg) tp.submit(scenario)说明上述配置类与函数签名以仓库中 taipy/common/config/config.pyi 与 taipy/core/config/ 为准具体的函数签名参数请以对应.pyi声明为准。五、公共 API场景、提交与查询__init__.py指出配置完成后只需import taipy as tp即可使用taipy顶层模块中的全部功能。最常用的函数包括tp.create_scenario()、tp.get_scenarios()、tp.get_data_nodes()、tp.submit()。这些函数全部实现在 taipy/core/taipy.py 中并最终委托给对应的 Manager 工厂。5.1 场景创建与 What-if 分析tp.create_scenario(config, creation_dateNone, nameNone)基于场景配置创建新场景会校验并锁定配置、管理应用版本若场景属于某个周期还会按创建日期与配置的 frequency 自动创建对应周期见 taipy/core/taipy.py。tp.get_scenarios(cycleNone, tagNone, is_sortedFalse, descendingFalse, created_start_timeNone, created_end_timeNone, sort_keyname)按周期、标签、创建时间过滤并排序场景排序键支持name、id、config_id、creation_date、tags见 taipy/core/taipy.py。tp.set_primary(scenario)/tp.get_primary(cycle)将某场景提升为该周期的主场景或将主场景降级get_primary_scenarios()可获取所有周期的主场景见 taipy/core/taipy.py。tp.compare_scenarios(*scenarios, data_node_config_idNone)对多个场景的数据节点做对比仅对配置过 comparator 的数据节点生效是 What-if 分析的重要入口见 taipy/core/taipy.py。tp.duplicate_scenario(scenario, ...)复制已有场景生成新场景可通过data_to_duplicate控制数据节点是否一并复制见 taipy/core/taipy.py。注意源码 docstring 明确警告目前只有基于文件的数据节点才能复制数据SQL、MongoDB 等数据节点复制后仍指向同一份数据需手动调整以避免冲突。5.2 提交与执行tp.submit(entity, forceFalse, waitFalse, timeoutNone, **properties)提交 Scenario、Sequence 或 Task 执行提交 Sequence/Scenario 时其下所有任务都会进入调度forceTrue可强制跳过本可跳过的任务wait/timeout用于同步等待执行完成见 taipy/core/taipy.py。返回Submission对象。tp.get_submissions()/tp.get_latest_submission(entity)/tp.get_jobs()/tp.get_latest_job(task)查询提交与作业记录见 taipy/core/taipy.py。tp.cancel_job(job)取消作业并将其后的作业置为ABANDONED状态见 taipy/core/taipy.py。tp.delete_job(job, forceFalse)/tp.delete_jobs()/tp.delete(entity_id)删除作业与各类实体见 taipy/core/taipy.py。5.3 实体查询与订阅tp.get(entity_id)按 ID 前缀自动识别实体类型并返回对应实体Task / DataNode / Sequence / Scenario / Job / Cycle / Submission见 taipy/core/taipy.py。tp.exists(entity_id)/tp.get_data_nodes()/tp.get_tasks()/tp.get_sequences()/tp.get_cycles()各类存在性与列表查询。tp.subscribe_scenario(callback, paramsNone, scenarioNone)/tp.subscribe_sequence(...)订阅 Job 状态变化回调用于在作业完成时触发通知或下游处理订阅仅对之后创建的作业生效见 taipy/core/taipy.py。tp.get_parents(entity)递归查找一个 Data Node / Task / Sequence 的全部父级实体结果按类型分组scenario、sequence、task见 taipy/core/taipy.py。tp.is_submittable(entity)、tp.is_editable(entity)、tp.is_readable(entity)、tp.is_deletable(entity)返回ReasonCollection对象可当布尔值使用并携带不可执行/不可编辑的具体原因见 taipy/core/taipy.py。tp.clean_all_entities(version_number)按版本号清空该版本下的全部实体Job、Submission、Scenario、Cycle、Sequence、Task、DataNode并删除版本见 taipy/core/taipy.py。六、Orchestrator 服务作业调度的运行引擎package_desc.md对应文档的__init__.py特别强调了 Orchestrator 服务Taipy provides a runnable service,Orchestrator^that runs as a service in a dedicated thread. The purpose is to have a dedicated thread responsible for dispatching the submitted jobs to an available executor for their execution. In particular, thisOrchestrator^service is automatically run when used with Taipy REST or Taipy GUI.在 taipy/core/orchestrator.py 中可以看到Orchestrator类的完整实现Orchestrator().run(force_restartFalse)启动服务。它会依次执行配置更新从命令行参数合并、版本管理_VersionManagerFactory、配置校验与配置锁定Config.block_update()最后启动 Job Dispatcher见 orchestrator.py。Orchestrator().stop(waitTrue, timeoutNone)停止 Dispatcher 并解锁配置Config.unblock_update()见 orchestrator.py。配置锁定意味着服务运行期间 Config 不再允许修改因此在调用tp.create_scenario()、tp.submit()等会触发版本管理与配置锁定的函数后后续的配置变更需要谨慎规划。旧版的Core服务类已废弃Core()现在会直接实例化并返回Orchestrator同时输出弃用警告见 taipy/core/_core.py。在开发模式Config.job_config.is_development下Orchestrator 启动时会同步检查并执行处于待运行状态的作业见 orchestrator.py。与 GUI/REST 集成时Orchestrator 会自动运行因此纯后端脚本模式下你需要自行启动服务才能让提交的任务真正被调度执行——否则提交的实体不会被执行源码中的submit函数也会发出对应警告见 taipy/core/taipy.py 上的装饰器_warn_no_orchestrator_service。七、版本管理版本管理是 taipy-core 面向生产运维的核心能力之一实现集中在 taipy/core/_version/_VersionManager见 taipy/core/_version/_version_manager.py负责版本的创建、切换、查询与删除所有实体数据节点、任务、场景等都按版本号隔离存储便于多版本实验并行与回滚上文提到的tp.clean_all_entities(version_number)提供了按版本批量清理实体的运维入口Orchestrator 启动时也会调用版本管理逻辑_manage_version()保证实体与当前代码版本一致。版本相关的迁移工具与 CLI 见 taipy/core/_version/_migrate_cli.py 与 taipy/core/_version/_cli/。八、运行测试与参与开发taipy/core/INSTALLATION.md 给出了开发者的测试流程使用 Pipenv 创建虚拟环境并安装开发依赖然后运行 taipy-core 的测试套件pip install pipenv pipenv install --dev pipenv run pytest tests/core当前仓库中的核心测试覆盖了数据节点读写如 tests/core/data/test_csv_data_node.py、tests/core/data/test_sql_data_node.py、场景管理tests/core/scenario/、作业与编排tests/core/_orchestrator/、版本管理tests/core/_version/等是理解各模块行为的最佳参考。九、许可证taipy-core 采用 Apache License 2.0 开源协议版权归 Avaiga Private LimitedCopyright 2021-2025许可全文可在 taipy/core/package_desc.md 与 LICENSE 中查看。协议明确允许使用、修改与再分发但需保留版权声明并遵守 Apache 2.0 的条款与条件。结语从数据节点到场景从提交执行到版本管理taipy-core 把业务应用后端最常见的诉求数据接入、任务编排、What-if 实验、多版本并存沉淀为一套结构清晰、可配置、可编程的 Python API。本文所涉及的源码路径taipy/core/taipy.py、taipy/core/orchestrator.py、taipy/core/config/、taipy/core/data/ 等均可在当前仓库中直接查阅进一步深入时可以结合 tests/core/ 下的测试用例验证各 API 的实际行为。【免费下载链接】taipyTurns Data and AI algorithms into production-ready web applications in no time.项目地址: https://gitcode.com/GitHub_Trending/ta/taipy创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考