开发工具CLI数据工程【免费下载链接】papermill Parameterize, execute, and analyze notebooks项目地址https://gitcode.com/gh_mirrors/pa/papermill点击查看免费下载Papermill 在开箱即用地支持本地文件、S3、GCS、HDFS 等多种读写来源和默认的 nbclient 本地执行引擎之外还提供了一套基于 Python Entry Points 的插件机制允许开发者在不修改 papermill 源码的前提下为其接入任意存储后端如 SFTP 服务器或自定义执行逻辑如远程执行、逐单元格计时统计。本文将以仓库docs/extending-overview.rst、docs/extending-entry-points.rst为核心脉络结合papermill/iorw.py、papermill/engines.py等源码实现完整讲解 I/O Handler 与 Engine 的接口约定、entry point 注册方式并给出两个可复制运行的完整实战示例使读者能够独立为 papermill 开发并交付自己的插件。扩展机制总览Papermill 的插件化设计用 papermill 运行一个 notebook 时背后其实只发生四件事见 docs/extending-overview.rst读取 notebook 文件将文件内容转换为 notebook 的 Python 对象nbformat 的 NotebookNode执行该 notebook将执行后的 notebook 写回文件。其中步骤 1、3、4 正是扩展点通过 entry points你可以编写自己的工具来接管读取I/O Handler、执行Engine与写回I/O Handler。步骤 2 由 papermill 内部完成通常无需干预。所谓 entry points是 Python 打包规范中已安装的分发包向外界通告自身组件供其他代码发现和使用的机制——典型如console_scripts生成命令行包装器以及 Pygments 通过它加载第三方语法高亮插件。Papermill 在运行时正是通过entrypoints库扫描两类 entry point 组papermill.io实现输入/输出I/O处理的 handlerpapermill.engine实现执行逻辑的 engine。从源码可以印证这一点papermill/iorw.py 中的PapermillIO.register_entry_points()调用entrypoints.get_group_all(papermill.io)并逐个注册papermill/engines.py 中的PapermillEngines.register_entry_points()则调用entrypoints.get_group_all(papermill.engine)。两个注册方法都在模块导入时被自动执行iorw.py 与 engines.py因此第三方插件只要以正常方式安装到环境中papermill 启动时即会自动发现。如果觉得仅靠新增 handler 和 engine 仍不够而是想改进项目本身的一些根本性设计则可以参与 papermill 的贡献开发详见下文参与 papermill 核心开发一节对应 docs/extending-developing.rst。开发新的 I/O HandlerI/O Handler 的接口约定四个必须实现的方法Papermill 中读取输入 notebook由 I/O Handler 管理正是它们让 papermill 不仅能访问本地文件系统还能访问 S3 等远程服务同样将执行后的 notebook 写回也由 I/O Handler 负责。因此I/O Handler 是 papermill 与任意存储后端对接的统一抽象层。编写自己的 I/O Handler就是编写一个实现了以下四个类方法的类CustomIO仅为示意名方法签名作用readCustomIO.read(file_path)返回文件内容字符串writeCustomIO.write(file_content, file_path)写入文件无返回值pretty_pathCustomIO.pretty_path(path)返回美化后的路径用于日志与展示listdirCustomIO.listdir(path)返回路径列表用于目录浏览原文档特别提醒如果你的 handler 只用于写例如只面向发布平台不打算支持read等操作那么应当实现该方法并在被调用时抛出异常如NotImplementedError而不是省略它——这样接口保持完整行为上又明确表达了此能力不受支持。这一约定与仓库内置 handler 的实现完全一致。例如 papermill/iorw.py 的LocalHandler实现了全部四个方法而只读性质更强的 handler 则主动对不支持的方法抛异常如GithubHandler.write抛PapermillException(write is not supported by GithubHandler)iorw.py、StreamHandler.listdir抛PapermillException(listdir is not supported by Stream Handler)iorw.py。这些实现可以作为编写部分能力受限 handler 的现成参考。确保 handler 被 papermill 发现pyproject.toml 注册开发完 handler 类之后需要在插件的pyproject.toml中声明 papermill 的 entry point。方法是在文件中加入[project.entry-points.papermill.io]段[project.entry-points.papermill.io] sftp:// papermill_sftp:SFTPHandler这一行的语义是当传入的文件路径以sftp://开头时papermill 就使用papermill_sftp包中导入的SFTPHandler类来处理该路径的读写。等号左边是路径前缀等号右边是类名及其导入来源格式为包名:类名。从源码看这套匹配逻辑实现在PapermillIO.get_handler中papermill/iorw.py它按注册顺序遍历self._handlers一旦path.startswith(scheme)命中即返回对应 handler若全部未命中则回退到localhandler即本地文件系统连本地 handler 都没有时才抛出PapermillException。另外要注意register是 LIFO 顺序插入iorw.py后注册的 scheme 会优先匹配设计同名前缀覆盖时需留意这一点。传统上papermill I/O handler 的 entry point 名采用 URL 前缀形式。仓库内置注册的 handler注意这些是 papermill 内部注册并非通过 entry point 加载但接口与约定相同包括见 papermill/iorw.pylocal→LocalHandler本地文件系统s3://→S3HandlerAmazon S3adl://→ADLHandlerAzure Data Lakeabs://→ABSHandlerAzure Blob Storagehttp://、https://→HttpHandlerHTTP/HTTPS 读写gs://→GCSHandlerGoogle Cloud Storage写入时带限流重试hdfs://→HDFSHandlerHadoop 文件系统http://github.com/、https://github.com/→GithubHandlerGitHub 内容读取-→StreamHandler标准输入/输出流。可以推断任何以 URL 风格前缀命名的新 handler 都能与这套体系自然共存。实战示例完整的 SFTP I/O Handler下面按原文档的演示从零构建一个可读写 SFTP 服务器的 handler目标是支持这样的命令行用法papermill sftp://my_ftp_server.co.uk/input.ipynb sftp://my_ftp_server.co.uk/output.ipynb项目结构如下papermill_sftp |- pyproject.toml |- src |- papermill_sftp |- __init__.py在src/papermill_sftp/__init__.py中实现 handler读取时先把远端文件下载到临时目录再读入写入时先写到临时文件再上传pretty_path直接原样返回路径listdir暂不实现按接口约定抛异常。原文档示例代码中未显式导入pathlib、tempfile、urllib.parse且cnopts需按你的 pysftp 环境配置这里补齐 import 并给出说明使代码可直接运行import os import pathlib import tempfile import urllib.parse import pysftp sftp_username os.getenv(SFTP_USERNAME) sftp_password os.getenv(SFTP_PASSWORD) # 根据你的环境配置主机密钥校验选项例如 # cnopts pysftp.CnOpts() # cnopts.hostkeys None # 仅测试环境使用生产环境请校验 host key cnopts pysftp.CnOpts() class SFTPHandler: classmethod def read(cls, path): Read a notebook from an SFTP server. parsed_url urllib.parse.urlparse(path) with tempfile.TemporaryDirectory() as tmpdir: tmp_file pathlib.Path(tmpdir) / pathlib.Path(parsed_url.path).name with pysftp.Connection( parsed_url.hostname, usernamesftp_username, passwordsftp_password, port(parsed_url.port or 22), cnoptscnopts, ) as sftp: sftp.get(parsed_url.path, str(tmp_file)) return tmp_file.read_text() classmethod def write(cls, file_content, path): Write a notebook to an SFTP server. parsed_url urllib.parse.urlparse(path) with tempfile.TemporaryDirectory() as tmpdir: tmp_file pathlib.Path(tmpdir) / output.ipynb tmp_file.write_text(file_content) with pysftp.Connection( parsed_url.hostname, usernamesftp_username, passwordsftp_password, port(parsed_url.port or 22), cnoptscnopts, ) as sftp: sftp.put(str(tmp_file), parsed_url.path) classmethod def pretty_path(cls, path): return path classmethod def listdir(cls, path): raise NotImplementedError配套的pyproject.toml完整内容如下注意[project.entry-points.papermill.io]段以及 setuptools 的 src 布局声明[build-system] requires [setuptools61.0, wheel] build-backend setuptools.build_meta [project] name papermill_sftp version 0.1 description An SFTP I/O handler for papermill. authors [ {name My Name, email my.emailgmail.com} ] dependencies [pysftp] [project.urls] Repository https://github.com/my_username/papermill_sftp.git [project.entry-points.papermill.io] sftp:// papermill_sftp:SFTPHandler [tool.setuptools] packages [papermill_sftp] package-dir { src}安装该插件后papermill 执行时会检查输入、输出路径是否以sftp://开头若命中则调用papermill_sftp中的SFTPHandler完成读与写。整个交互链路为sftp://路径 →PapermillIO.get_handler前缀匹配iorw.py→SFTPHandler.read/write→ 继续走 papermill 的读取/写回流程。开发新的执行引擎EngineEngine 基类与 execute_managed_notebook 接口Papermill 的 engine 是能够执行一个 notebook 的 Python 对象。默认实现NBClientEngine接收一个 notebook 对象并在本机执行其背后是 nbclient 的PapermillNotebookClient见 papermill/engines.py。通过编写自定义 engine你可以把执行交给远程服务器或在执行后对 notebook 做后处理例如注入额外的输出单元格。自定义 engine 需要继承papermill.engines.Engine基类并实现类方法execute_managed_notebook其调用签名要与父类保持一致class CustomEngine(papermill.engines.Engine): classmethod def execute_managed_notebook(cls, nb_man, kernel_name, **kwargs): pass这里需要澄清一个容易误解的细节原文档将nb_man描述为nbformat.NotebookNode对象但从源码看papermill/engines.pyEngine.execute_notebook会先把 notebook 包装进NotebookExecutionManager其内部nb属性才是NotebookNode再把该管理器实例传给execute_managed_notebook。NotebookExecutionManager封装了统一的执行状态管理notebook_start初始化并清空 papermill 元数据、cell_start/cell_exception/cell_complete逐单元格更新状态与耗时、notebook_complete收尾并强制保存还内置了进度条与自动保存autosave_cell_every默认 30 秒见 engines.py。因此自定义 engine 只需聚焦如何逐单元格执行而元数据记录、保存等横切逻辑可全部复用。基类Engine的默认execute_managed_notebook直接抛出NotImplementedErrorengines.py强制子类实现。实战示例记录每个代码单元格耗时的 Timing Engine原文档提供了一个完整的演示实现一个自定义 engine把每个代码单元格的执行耗时作为额外输出注入到该单元格的 outputs 开头。因为 papermill 本身已在单元格元数据中记录start_time/end_time我们直接复用默认引擎NBClientEngine再借助nbformat的new_output构造输出节点。项目结构papermill_timing |- pyproject.toml |- src |- papermill_timing |- __init__.pysrc/papermill_timing/__init__.py内容如下from datetime import datetime from papermill.engines import NBClientEngine from nbformat.v4 import new_output class CustomEngine(NBClientEngine): classmethod def execute_managed_notebook(cls, nb_man, kernel_name, **kwargs): # call the papermill execution engine: super().execute_managed_notebook(nb_man, kernel_name, **kwargs) for cell in nb_man.nb.cells: if cell.cell_type code and cell.execution_count is not None: start datetime.fromisoformat(cell.metadata.papermill.start_time) end datetime.fromisoformat(cell.metadata.papermill.end_time) output_message fExecution took {(end - start).total_seconds():.3f} seconds output_node new_output(display_data, data{text/plain: [output_message]}) cell.outputs [output_node] cell.outputs实现要点先调用super().execute_managed_notebook(...)走完默认的本地执行流程此时cell.metadata.papermill.start_time/end_time已被NotebookExecutionManager.cell_start/cell_complete写入这两个回调分别见 engines.py 与 engines.py随后遍历代码单元格计算耗时差并把新构造的display_data输出节点插到原有 outputs 之前。确保 engine 被 papermill 发现pyproject.toml 注册自定义 engine 需要以papermill.engine为前缀注册为 entry point引用我们刚实现的类。以papermill_timing包中的CustomEngine为例[build-system] requires [setuptools61.0, wheel] build-backend setuptools.build_meta [project] name papermill_timing version 0.1 description A papermill engine that logs additional timing information about code. authors [ {name My Name, email my.emailgmail.com} ] dependencies [papermill, nbformat] [project.urls] Repository https://github.com/my_username/papermill_timing.git [project.entry-points.papermill.engine] timer_engine papermill_timing:CustomEngine [tool.setuptools] packages [papermill_timing] package-dir { src}注册后用户即可通过命令行参数--engine timer_engine选用该引擎该参数在 papermill/cli.py 中定义为执行 notebook 时使用的引擎名称papermill input.ipynb output.ipynb --engine timer_enginePapermillEngines.get_engine会按名称在已注册的引擎字典中查找找不到时抛出PapermillException(fNo engine named {name} found)engines.py。仓库内置注册了None与nbclient两个名称均指向NBClientEngineengines.py自定义名称与内置名称互不冲突。下图左侧为使用自定义 timing engine 执行后的 notebook右侧为使用标准 engine 执行的结果每个代码单元格顶部都多出了我们注入的耗时输出配图源自 docs/extending-entry-points.rst关于 nb_man 回调的更轻量写法如果你的自定义 engine 不想走完整的 nbclient 执行流程而只是想在默认流程之外做点事情可以像测试用例CellCallbackEngine那样直接操纵回调在 papermill/tests/test_engines.py 中一个继承Engine的最小实现只调用nb_man.cell_start(cell)与nb_man.cell_complete(cell)配合Engine.execute_notebook的包装即可完成元数据的完整记录与保存。这验证了引擎实现者的核心职责只是决定什么时候执行、执行什么状态机与持久化由NotebookExecutionManager统一兜底。参与 papermill 核心开发当扩展需求超出新增 I/O handler 与执行 handler的范畴涉及项目根本性改进时可以选择直接向 papermill 贡献代码详见 docs/extending-developing.rst。仓库内已提供 CONTRIBUTING.md、CODE_OF_CONDUCT.md 与 DEVELOPMENT_GUIDE.md 供贡献者查阅开始之前建议先通读。开发过程中papermill/tests/下的测试套件例如 test_engines.py 中的TestEngineRegistration、test_iorw.py 中的test_entrypoint_register分别验证了 engine 与 I/O handler 的 entry point 注册逻辑可作为自己插件行为对标的回归参考。小结Papermill 的扩展体系可以概括为两条清晰的插件通道I/O Handlerpapermill.io组实现read/write/pretty_path/listdir四个类方法以 URL 前缀注册即可让 papermill 读写任意存储后端默认回退到本地文件系统Enginepapermill.engine组继承papermill.engines.Engine并实现execute_managed_notebook通过--engine 名称在命令行选用可接管或增强执行逻辑远程执行、结果后处理、重复运行模拟等。两者都只需在插件自己的pyproject.toml中声明 entry point、正常安装即可被 papermill 自动发现完全不需要改动 papermill 本体。理解这些接口约定以及NotebookExecutionManager的状态管理语义之后无论是接入私有存储、对接远程执行集群还是为 notebook 注入自定义分析输出都可以在数十分钟内落地为一个独立的可复用插件包。赞分享开发工具CLI数据工程【免费下载链接】papermill Parameterize, execute, and analyze notebooks项目地址https://gitcode.com/gh_mirrors/pa/papermill点击查看免费下载相关推荐解锁AI创作新维度5个关键策略让您的ComfyUI多GPU效率提升3倍解锁AI创作新维度5个关键策略让您的ComfyUI多GPU效率提升3倍 您是否曾因显卡显存不足而无法运行心仪的大型AI模型是否在生成高分辨率图像或视频时频繁人工智能大模型深度学习本地部署终极指南如何扩展Papermill自定义引擎与翻译器开发终极指南如何扩展Papermill自定义引擎与翻译器开发 想要让Papermill参数化执行工具发挥更大威力吗这篇完整教程将教你如何通过自定义引擎和翻开发工具CLI数据工程Woodpecker 自定义 Backend 开发指南通过 Backend 接口与 RunAgent 扩展 CI/CD 执行引擎Woodpecker 自定义 Backend 开发指南通过 Backend 接口与 RunAgent 扩展 CI/CD 执行引擎 本文面向需要为 WoodpeCI/CDDevOps上一篇PPT Master把手头的文档快速变成原生可编辑的 PPT下一篇Flowframes本地视频插帧指南把24fps素材输出成60fps流畅画面创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Papermill 扩展开发实战：通过 Entry Points 自定义 I/O 处理器与执行引擎
开发工具CLI数据工程【免费下载链接】papermill Parameterize, execute, and analyze notebooks项目地址https://gitcode.com/gh_mirrors/pa/papermill点击查看免费下载Papermill 在开箱即用地支持本地文件、S3、GCS、HDFS 等多种读写来源和默认的 nbclient 本地执行引擎之外还提供了一套基于 Python Entry Points 的插件机制允许开发者在不修改 papermill 源码的前提下为其接入任意存储后端如 SFTP 服务器或自定义执行逻辑如远程执行、逐单元格计时统计。本文将以仓库docs/extending-overview.rst、docs/extending-entry-points.rst为核心脉络结合papermill/iorw.py、papermill/engines.py等源码实现完整讲解 I/O Handler 与 Engine 的接口约定、entry point 注册方式并给出两个可复制运行的完整实战示例使读者能够独立为 papermill 开发并交付自己的插件。扩展机制总览Papermill 的插件化设计用 papermill 运行一个 notebook 时背后其实只发生四件事见 docs/extending-overview.rst读取 notebook 文件将文件内容转换为 notebook 的 Python 对象nbformat 的 NotebookNode执行该 notebook将执行后的 notebook 写回文件。其中步骤 1、3、4 正是扩展点通过 entry points你可以编写自己的工具来接管读取I/O Handler、执行Engine与写回I/O Handler。步骤 2 由 papermill 内部完成通常无需干预。所谓 entry points是 Python 打包规范中已安装的分发包向外界通告自身组件供其他代码发现和使用的机制——典型如console_scripts生成命令行包装器以及 Pygments 通过它加载第三方语法高亮插件。Papermill 在运行时正是通过entrypoints库扫描两类 entry point 组papermill.io实现输入/输出I/O处理的 handlerpapermill.engine实现执行逻辑的 engine。从源码可以印证这一点papermill/iorw.py 中的PapermillIO.register_entry_points()调用entrypoints.get_group_all(papermill.io)并逐个注册papermill/engines.py 中的PapermillEngines.register_entry_points()则调用entrypoints.get_group_all(papermill.engine)。两个注册方法都在模块导入时被自动执行iorw.py 与 engines.py因此第三方插件只要以正常方式安装到环境中papermill 启动时即会自动发现。如果觉得仅靠新增 handler 和 engine 仍不够而是想改进项目本身的一些根本性设计则可以参与 papermill 的贡献开发详见下文参与 papermill 核心开发一节对应 docs/extending-developing.rst。开发新的 I/O HandlerI/O Handler 的接口约定四个必须实现的方法Papermill 中读取输入 notebook由 I/O Handler 管理正是它们让 papermill 不仅能访问本地文件系统还能访问 S3 等远程服务同样将执行后的 notebook 写回也由 I/O Handler 负责。因此I/O Handler 是 papermill 与任意存储后端对接的统一抽象层。编写自己的 I/O Handler就是编写一个实现了以下四个类方法的类CustomIO仅为示意名方法签名作用readCustomIO.read(file_path)返回文件内容字符串writeCustomIO.write(file_content, file_path)写入文件无返回值pretty_pathCustomIO.pretty_path(path)返回美化后的路径用于日志与展示listdirCustomIO.listdir(path)返回路径列表用于目录浏览原文档特别提醒如果你的 handler 只用于写例如只面向发布平台不打算支持read等操作那么应当实现该方法并在被调用时抛出异常如NotImplementedError而不是省略它——这样接口保持完整行为上又明确表达了此能力不受支持。这一约定与仓库内置 handler 的实现完全一致。例如 papermill/iorw.py 的LocalHandler实现了全部四个方法而只读性质更强的 handler 则主动对不支持的方法抛异常如GithubHandler.write抛PapermillException(write is not supported by GithubHandler)iorw.py、StreamHandler.listdir抛PapermillException(listdir is not supported by Stream Handler)iorw.py。这些实现可以作为编写部分能力受限 handler 的现成参考。确保 handler 被 papermill 发现pyproject.toml 注册开发完 handler 类之后需要在插件的pyproject.toml中声明 papermill 的 entry point。方法是在文件中加入[project.entry-points.papermill.io]段[project.entry-points.papermill.io] sftp:// papermill_sftp:SFTPHandler这一行的语义是当传入的文件路径以sftp://开头时papermill 就使用papermill_sftp包中导入的SFTPHandler类来处理该路径的读写。等号左边是路径前缀等号右边是类名及其导入来源格式为包名:类名。从源码看这套匹配逻辑实现在PapermillIO.get_handler中papermill/iorw.py它按注册顺序遍历self._handlers一旦path.startswith(scheme)命中即返回对应 handler若全部未命中则回退到localhandler即本地文件系统连本地 handler 都没有时才抛出PapermillException。另外要注意register是 LIFO 顺序插入iorw.py后注册的 scheme 会优先匹配设计同名前缀覆盖时需留意这一点。传统上papermill I/O handler 的 entry point 名采用 URL 前缀形式。仓库内置注册的 handler注意这些是 papermill 内部注册并非通过 entry point 加载但接口与约定相同包括见 papermill/iorw.pylocal→LocalHandler本地文件系统s3://→S3HandlerAmazon S3adl://→ADLHandlerAzure Data Lakeabs://→ABSHandlerAzure Blob Storagehttp://、https://→HttpHandlerHTTP/HTTPS 读写gs://→GCSHandlerGoogle Cloud Storage写入时带限流重试hdfs://→HDFSHandlerHadoop 文件系统http://github.com/、https://github.com/→GithubHandlerGitHub 内容读取-→StreamHandler标准输入/输出流。可以推断任何以 URL 风格前缀命名的新 handler 都能与这套体系自然共存。实战示例完整的 SFTP I/O Handler下面按原文档的演示从零构建一个可读写 SFTP 服务器的 handler目标是支持这样的命令行用法papermill sftp://my_ftp_server.co.uk/input.ipynb sftp://my_ftp_server.co.uk/output.ipynb项目结构如下papermill_sftp |- pyproject.toml |- src |- papermill_sftp |- __init__.py在src/papermill_sftp/__init__.py中实现 handler读取时先把远端文件下载到临时目录再读入写入时先写到临时文件再上传pretty_path直接原样返回路径listdir暂不实现按接口约定抛异常。原文档示例代码中未显式导入pathlib、tempfile、urllib.parse且cnopts需按你的 pysftp 环境配置这里补齐 import 并给出说明使代码可直接运行import os import pathlib import tempfile import urllib.parse import pysftp sftp_username os.getenv(SFTP_USERNAME) sftp_password os.getenv(SFTP_PASSWORD) # 根据你的环境配置主机密钥校验选项例如 # cnopts pysftp.CnOpts() # cnopts.hostkeys None # 仅测试环境使用生产环境请校验 host key cnopts pysftp.CnOpts() class SFTPHandler: classmethod def read(cls, path): Read a notebook from an SFTP server. parsed_url urllib.parse.urlparse(path) with tempfile.TemporaryDirectory() as tmpdir: tmp_file pathlib.Path(tmpdir) / pathlib.Path(parsed_url.path).name with pysftp.Connection( parsed_url.hostname, usernamesftp_username, passwordsftp_password, port(parsed_url.port or 22), cnoptscnopts, ) as sftp: sftp.get(parsed_url.path, str(tmp_file)) return tmp_file.read_text() classmethod def write(cls, file_content, path): Write a notebook to an SFTP server. parsed_url urllib.parse.urlparse(path) with tempfile.TemporaryDirectory() as tmpdir: tmp_file pathlib.Path(tmpdir) / output.ipynb tmp_file.write_text(file_content) with pysftp.Connection( parsed_url.hostname, usernamesftp_username, passwordsftp_password, port(parsed_url.port or 22), cnoptscnopts, ) as sftp: sftp.put(str(tmp_file), parsed_url.path) classmethod def pretty_path(cls, path): return path classmethod def listdir(cls, path): raise NotImplementedError配套的pyproject.toml完整内容如下注意[project.entry-points.papermill.io]段以及 setuptools 的 src 布局声明[build-system] requires [setuptools61.0, wheel] build-backend setuptools.build_meta [project] name papermill_sftp version 0.1 description An SFTP I/O handler for papermill. authors [ {name My Name, email my.emailgmail.com} ] dependencies [pysftp] [project.urls] Repository https://github.com/my_username/papermill_sftp.git [project.entry-points.papermill.io] sftp:// papermill_sftp:SFTPHandler [tool.setuptools] packages [papermill_sftp] package-dir { src}安装该插件后papermill 执行时会检查输入、输出路径是否以sftp://开头若命中则调用papermill_sftp中的SFTPHandler完成读与写。整个交互链路为sftp://路径 →PapermillIO.get_handler前缀匹配iorw.py→SFTPHandler.read/write→ 继续走 papermill 的读取/写回流程。开发新的执行引擎EngineEngine 基类与 execute_managed_notebook 接口Papermill 的 engine 是能够执行一个 notebook 的 Python 对象。默认实现NBClientEngine接收一个 notebook 对象并在本机执行其背后是 nbclient 的PapermillNotebookClient见 papermill/engines.py。通过编写自定义 engine你可以把执行交给远程服务器或在执行后对 notebook 做后处理例如注入额外的输出单元格。自定义 engine 需要继承papermill.engines.Engine基类并实现类方法execute_managed_notebook其调用签名要与父类保持一致class CustomEngine(papermill.engines.Engine): classmethod def execute_managed_notebook(cls, nb_man, kernel_name, **kwargs): pass这里需要澄清一个容易误解的细节原文档将nb_man描述为nbformat.NotebookNode对象但从源码看papermill/engines.pyEngine.execute_notebook会先把 notebook 包装进NotebookExecutionManager其内部nb属性才是NotebookNode再把该管理器实例传给execute_managed_notebook。NotebookExecutionManager封装了统一的执行状态管理notebook_start初始化并清空 papermill 元数据、cell_start/cell_exception/cell_complete逐单元格更新状态与耗时、notebook_complete收尾并强制保存还内置了进度条与自动保存autosave_cell_every默认 30 秒见 engines.py。因此自定义 engine 只需聚焦如何逐单元格执行而元数据记录、保存等横切逻辑可全部复用。基类Engine的默认execute_managed_notebook直接抛出NotImplementedErrorengines.py强制子类实现。实战示例记录每个代码单元格耗时的 Timing Engine原文档提供了一个完整的演示实现一个自定义 engine把每个代码单元格的执行耗时作为额外输出注入到该单元格的 outputs 开头。因为 papermill 本身已在单元格元数据中记录start_time/end_time我们直接复用默认引擎NBClientEngine再借助nbformat的new_output构造输出节点。项目结构papermill_timing |- pyproject.toml |- src |- papermill_timing |- __init__.pysrc/papermill_timing/__init__.py内容如下from datetime import datetime from papermill.engines import NBClientEngine from nbformat.v4 import new_output class CustomEngine(NBClientEngine): classmethod def execute_managed_notebook(cls, nb_man, kernel_name, **kwargs): # call the papermill execution engine: super().execute_managed_notebook(nb_man, kernel_name, **kwargs) for cell in nb_man.nb.cells: if cell.cell_type code and cell.execution_count is not None: start datetime.fromisoformat(cell.metadata.papermill.start_time) end datetime.fromisoformat(cell.metadata.papermill.end_time) output_message fExecution took {(end - start).total_seconds():.3f} seconds output_node new_output(display_data, data{text/plain: [output_message]}) cell.outputs [output_node] cell.outputs实现要点先调用super().execute_managed_notebook(...)走完默认的本地执行流程此时cell.metadata.papermill.start_time/end_time已被NotebookExecutionManager.cell_start/cell_complete写入这两个回调分别见 engines.py 与 engines.py随后遍历代码单元格计算耗时差并把新构造的display_data输出节点插到原有 outputs 之前。确保 engine 被 papermill 发现pyproject.toml 注册自定义 engine 需要以papermill.engine为前缀注册为 entry point引用我们刚实现的类。以papermill_timing包中的CustomEngine为例[build-system] requires [setuptools61.0, wheel] build-backend setuptools.build_meta [project] name papermill_timing version 0.1 description A papermill engine that logs additional timing information about code. authors [ {name My Name, email my.emailgmail.com} ] dependencies [papermill, nbformat] [project.urls] Repository https://github.com/my_username/papermill_timing.git [project.entry-points.papermill.engine] timer_engine papermill_timing:CustomEngine [tool.setuptools] packages [papermill_timing] package-dir { src}注册后用户即可通过命令行参数--engine timer_engine选用该引擎该参数在 papermill/cli.py 中定义为执行 notebook 时使用的引擎名称papermill input.ipynb output.ipynb --engine timer_enginePapermillEngines.get_engine会按名称在已注册的引擎字典中查找找不到时抛出PapermillException(fNo engine named {name} found)engines.py。仓库内置注册了None与nbclient两个名称均指向NBClientEngineengines.py自定义名称与内置名称互不冲突。下图左侧为使用自定义 timing engine 执行后的 notebook右侧为使用标准 engine 执行的结果每个代码单元格顶部都多出了我们注入的耗时输出配图源自 docs/extending-entry-points.rst关于 nb_man 回调的更轻量写法如果你的自定义 engine 不想走完整的 nbclient 执行流程而只是想在默认流程之外做点事情可以像测试用例CellCallbackEngine那样直接操纵回调在 papermill/tests/test_engines.py 中一个继承Engine的最小实现只调用nb_man.cell_start(cell)与nb_man.cell_complete(cell)配合Engine.execute_notebook的包装即可完成元数据的完整记录与保存。这验证了引擎实现者的核心职责只是决定什么时候执行、执行什么状态机与持久化由NotebookExecutionManager统一兜底。参与 papermill 核心开发当扩展需求超出新增 I/O handler 与执行 handler的范畴涉及项目根本性改进时可以选择直接向 papermill 贡献代码详见 docs/extending-developing.rst。仓库内已提供 CONTRIBUTING.md、CODE_OF_CONDUCT.md 与 DEVELOPMENT_GUIDE.md 供贡献者查阅开始之前建议先通读。开发过程中papermill/tests/下的测试套件例如 test_engines.py 中的TestEngineRegistration、test_iorw.py 中的test_entrypoint_register分别验证了 engine 与 I/O handler 的 entry point 注册逻辑可作为自己插件行为对标的回归参考。小结Papermill 的扩展体系可以概括为两条清晰的插件通道I/O Handlerpapermill.io组实现read/write/pretty_path/listdir四个类方法以 URL 前缀注册即可让 papermill 读写任意存储后端默认回退到本地文件系统Enginepapermill.engine组继承papermill.engines.Engine并实现execute_managed_notebook通过--engine 名称在命令行选用可接管或增强执行逻辑远程执行、结果后处理、重复运行模拟等。两者都只需在插件自己的pyproject.toml中声明 entry point、正常安装即可被 papermill 自动发现完全不需要改动 papermill 本体。理解这些接口约定以及NotebookExecutionManager的状态管理语义之后无论是接入私有存储、对接远程执行集群还是为 notebook 注入自定义分析输出都可以在数十分钟内落地为一个独立的可复用插件包。赞分享开发工具CLI数据工程【免费下载链接】papermill Parameterize, execute, and analyze notebooks项目地址https://gitcode.com/gh_mirrors/pa/papermill点击查看免费下载相关推荐解锁AI创作新维度5个关键策略让您的ComfyUI多GPU效率提升3倍解锁AI创作新维度5个关键策略让您的ComfyUI多GPU效率提升3倍 您是否曾因显卡显存不足而无法运行心仪的大型AI模型是否在生成高分辨率图像或视频时频繁人工智能大模型深度学习本地部署终极指南如何扩展Papermill自定义引擎与翻译器开发终极指南如何扩展Papermill自定义引擎与翻译器开发 想要让Papermill参数化执行工具发挥更大威力吗这篇完整教程将教你如何通过自定义引擎和翻开发工具CLI数据工程Woodpecker 自定义 Backend 开发指南通过 Backend 接口与 RunAgent 扩展 CI/CD 执行引擎Woodpecker 自定义 Backend 开发指南通过 Backend 接口与 RunAgent 扩展 CI/CD 执行引擎 本文面向需要为 WoodpeCI/CDDevOps上一篇PPT Master把手头的文档快速变成原生可编辑的 PPT下一篇Flowframes本地视频插帧指南把24fps素材输出成60fps流畅画面创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考