SFTPOperator throws object of type 'PlainXComArg' has no len() when using with Taskflow API

[jtommi]

Apache Airflow Provider(s)

sftp

Versions of Apache Airflow Providers

apache-airflow-providers-sftp==4.1.0

Apache Airflow version

2.4.2 Python 3.10

Operating System

Debian 11 (Official docker image)

Deployment

Docker-Compose

Deployment details

Base image is apache/airflow:2.4.2-python3.10

What happened

When combining Taskflow API and SFTPOperator, it throws an exception that didn't happen with apache-airflow-providers-sftp 4.0.0

What you think should happen instead

The DAG should work as expected

How to reproduce

import pendulum

from airflow import DAG
from airflow.decorators import task
from airflow.providers.sftp.operators.sftp import SFTPOperator

with DAG(
    "example_sftp",
    schedule="@once",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    catchup=False,
    tags=["example"],
) as dag:

    @task
    def get_file_path():
        return "test.csv"

    local_filepath = get_file_path()

    upload = SFTPOperator(
        task_id=f"upload_file_to_sftp",
        ssh_conn_id="sftp_connection",
        local_filepath=local_filepath,
        remote_filepath="test.csv",
    )

Anything else

[2022-10-27T15:21:38.106+0000]` {logging_mixin.py:120} INFO - [2022-10-27T15:21:38.102+0000] {dagbag.py:342} ERROR - Failed to import: /opt/airflow/dags/test.py
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/dagbag.py", line 338, in parse
    loader.exec_module(new_module)
  File "<frozen importlib._bootstrap_external>", line 883, in exec_module
  File "<frozen importlib._bootstrap>", line 241, in _call_with_frames_removed
  File "/opt/airflow/dags/test.py", line 21, in <module>
    upload = SFTPOperator(
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/models/baseoperator.py", line 408, in apply_defaults
    result = func(self, **kwargs, default_args=default_args)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/providers/sftp/operators/sftp.py", line 116, in __init__
    if len(self.local_filepath) != len(self.remote_filepath):
TypeError: object of type 'PlainXComArg' has no len()

It looks like the offending code was introduced in commit 5f073e3

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[o-nikolas]

Thanks for the bug report @jtommi!

I see that you've checked that you're willing to submit a PR, so I have assigned the task to you 😃

[uranusjr]

Value checks in operators should be done in execute, not __init__.

[jtommi]

@uranusjr Thanks for the initial direction to fixing this.

I drafted #27346.
I think the initial bug is fixed, but my test that checks for compatibility is failing.

airflow/tests/providers/sftp/operators/test_sftp.py

Line 413 in c848311

def test_local_remote_file_paths_compatible_with_taskflow(self, mock_get):

The reason is expected str, bytes or os.PathLike object, not PlainXComArg on

airflow/airflow/providers/sftp/operators/sftp.py

Line 173 in c848311

local_folder = os.path.dirname(local_filepath)

The thing is that I don't understand why the code worked with version 2 of the provider package (I actually didn't check with version 4, but I didn't get import errors).
The only change from previous versions of the code is that before it did os.path.dirname(self.local_filepath) and now os.path.dirname(local_filepath), but I couldn't find any code that converted a PlainXComArg into a string.

[uranusjr]

All the checks should be moved to execute anyway, just do that and don’t worry that much about the previous logic.

[jtommi]

The thing is: either I broke (or didn't find) the code that made Taskflow work in version 4.0 of this package, or my test is incorrect.

[potiuk]

This should also be fixed regardless of the SFTP provider version by #27251

[jtommi]

This should also be fixed regardless of the SFTP provider version by #27251

Great, I'll subscribe to #27251 and verify that it fixes the issue.
Thanks @potiuk for letting me know

[potiuk]

Yeah. It's been released in 2.5.0 so just upgrading to 2.5.0 should be enough to test it.