自定义Airflow SFTPOperator实现多文件SFTP下载目录访问报错排查

需求背景 #

• 原生Apache Airflow 1.10.12版本的SFTPOperator仅支持单文件单次传输，需要扩展实现从SFTP服务器批量筛选下载多文件的能力。
• 扩展逻辑的核心设计：遍历指定远程目录下的所有文件，仅筛选修改时间处于「前一天零点到当天零点」区间内的文件下载到本地。
• 实现方式：拷贝原算子源码做二次开发，新增get_xml_from_source()函数实现目录遍历、文件时间筛选、批量下载逻辑，仅替换原算子中单文件传输的核心代码，其余逻辑保持不变。

报错现象 #

• 任务运行时直接在files_attr = sftp_client.listdir_attr(remote_filepath)这行抛出错误，报错内容为「从/source/ 路径到/path/to/destination 路径传输时出错，错误：[Errno 2] 没有对应的文件或目录」，提示找不到指定的SFTP远程目录。
• 使用相同SFTP连接配置、相同函数逻辑在本地独立运行时，功能完全正常，未出现目录找不到的问题。
• 已排查过SFTP连接层的paramiko配置，未发现连接参数异常。

根因定位 #

所有问题都不是SFTP连接本身的权限限制，全是参数传值、逻辑写死、路径匹配的问题：

• 远程路径语义和算子设计不匹配
  原SFTPOperator的remote_filepath参数设计是「指向单个待传输文件的完整路径」，本身没有对路径是目录还是文件做校验。你在任务配置里传入的remote_filepath="/source/"是目录路径，两个隐藏坑直接触发报错：
  1. Airflow底层SSHHook建立SFTP会话时，初始工作目录是对应用户的家目录，和你本地测试时用paramiko直连后的工作目录没有做对齐校验——哪怕路径写的是绝对路径，SFTP服务端对路径的权限映射、家目录挂载逻辑可能存在差异，本地能识别的路径不代表Airflow任务执行时的会话能正常访问。
  2. 遍历拿到所有匹配文件后的下载逻辑存在硬伤：循环里写的是sftp_client.get(remote_filepath, local_filepath)，根本没拼接单个文件的全路径，哪怕目录能正常遍历，实际下载时传的还是整个目录的路径，完全没用到循环里遍历出来的file_name变量，和原单文件算子的传参逻辑完全不兼容。
• 本地路径拆分逻辑错误
  原算子里用os.path.dirname(self.local_filepath)取本地目标目录的逻辑，是建立在「local_filepath是带具体文件名的完整本地文件路径」的前提上的。你在任务配置里传入的local_filepath=f"{current_dir}/destination"本身是本地目标目录，没有带具体文件名，用os.path.dirname取出来的路径会直接上跳一级目录——比如传入路径是/opt/dags/destination，取出来的目录是/opt/dags，根本没把destination目录纳入创建范围，哪怕远程目录能访问，本地路径也会因为层级错误触发报错。
• 时间筛选逻辑硬编码
  你在函数里硬编码了用datetime.today()取当天零点、前一天零点做筛选，完全没用上传入的prev_execution_date、execution_date参数，哪怕路径问题解决了，遇到任务补跑、延迟调度的场景，时间筛选逻辑也会和任务计划完全错位，漏下载符合要求的文件。

修复方案 #

• 第一步：对齐远程路径校验
  1. 建立SFTP连接后，先执行sftp_client.getcwd()打印当前会话的工作目录，和本地测试时的工作目录做比对，确认是否存在SFTP服务端的路径映射、权限差异，再用sftp_client.stat(remote_filepath)校验路径是否真实存在，减少盲报的目录不存在问题。
  2. 遍历目录时，先去掉remote_filepath末尾的斜杠，逐文件拼接完整的远程文件全路径，不要直接把目录路径传给sftp_client.get()方法。
• 第二步：修正本地路径处理逻辑
  废弃原算子里自动取本地路径父层级的逻辑，直接判断传入的local_filepath对应的本地路径是否是已存在的目录，如果是目录，就把远程遍历拿到的文件名拼在这个目录路径后面，作为本地单文件的完整存储路径，不要把传入的本地目录路径硬当具体文件路径处理。
• 第三步：替换硬编码的时间逻辑
  删掉原来用datetime.today()写死的时间筛选代码，直接用任务通过Jinja模板渲染传入的prev_execution_date、execution_date作为筛选区间，确保任务补跑、延迟调度时的文件筛选逻辑和任务触发时间完全对齐。
• 第四步：修正循环内的下载逻辑
  把循环里的下载代码改成逐文件拼接路径的形式：

  full_remote_file = os.path.join(remote_filepath.rstrip('/'), file_name)
  target_local_file = os.path.join(local_filepath.rstrip('/'), file_name)
  sftp_client.get(full_remote_file, target_local_file)
  

  原代码在循环里不管匹配到多少个文件，永远下载的是外层传入的目录路径，根本没有逐文件读取遍历结果，这是哪怕目录遍历成功，后续逻辑也完全跑不通的核心bug。

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