GCSToGCSOperator source_objects failing to parse xcom

[blaklaybul]

Apache Airflow Provider(s)

google

Versions of Apache Airflow Providers

10.15.0

Apache Airflow version

2.7.3

Operating System

debian

Deployment

Astronomer

Deployment details

Using astro runtime 9.15.0

What happened

The GCSToGCSOperator is throwing an error when trying to pass task output into the source_objects field.

 File "/usr/local/lib/python3.11/site-packages/airflow/providers/google/cloud/transfers/gcs_to_gcs.py", line 211, in __init__
    if source_objects and any(WILDCARD in obj for obj in source_objects):
                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/airflow/models/xcom_arg.py", line 268, in __iter__
    raise TypeError("'XComArg' object is not iterable")

What you think should happen instead

This operator should be able to handle xcom, and we have no issue using the GCSToBigQueryOperator in the same way

How to reproduce

@task
def return_files():
    files = ['a','b','c','d']
    return files

files = return_files()

 extract_files = GCSToGCSOperator(
        task_id="fetch_data",
        source_bucket="source_bucket_name",
        source_objects=files,
        destination_bucket="dest_bucket_name",
        destination_object="foo",
        dag=dag,
    )

files >> extract_files

Anything else

No response

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

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

You are not supposed to pass task results (which are dynamic) as initialization parameters in classic operator. This is not going to work - in classic operators the only way to get parameters dynamically resolved during execution is to use jinja templates.

If you want to pass the list, you need to pass it as jinja template where you pull parameters from xcom and use "render as native object". Or you can write what GCS to GCS operator does using hooks in taskflow-annotated task similarly to what is described in https://medium.com/apache-airflow/generic-airflow-transfers-made-easy-5fe8e5e7d2c2

[judoole]

Sorry for coming in late here @potiuk, with classic operator, does that mean all operators that are created the original way with constructor, inheritance from BaseOperator etc?
We use the pattern of dynamic results a lot actually, because it automatically adds tasks as upstreams. Thus, we have a self maintained way of keeping the graph and task dependencies up to date, and the graph becomes a better documentation of the actual dependencies.

Official operators that is lacking templated fields, or is doing constructor verification on templated fields as str, we rewrite so that they work. For example the GCSToGCSOperator is overwritten as so;

from airflow.providers.google.cloud.transfers.gcs_to_gcs import GCSToGCSOperator as BaseGCSToGCSOperator

class GCSToGCSOperator(BaseGCSToGCSOperator):
    def __init__(
        self,
        *,
        source_bucket,
        source_object=None,
        source_objects=None,
        destination_bucket=None,
        destination_object=None,
        delimiter=None,
        move_object=False,
        replace=True,
        gcp_conn_id="google_cloud_default",
        last_modified_time=None,
        maximum_modified_time=None,
        is_older_than=None,
        impersonation_chain: str | Sequence[str] | None = None,
        source_object_required=False,
        exact_match=False,
        match_glob: str | None = None,
        **kwargs,
    ):
        """Because the GCSToGCSOperator cannot handle templated source objects, we need to override the constructor"""
        # First init constructor of the BaseOperator
        BaseOperator.__init__(self, **kwargs)
        # Next set all fields like in the superclass constructor
        # Skipping the checks for wildcard in source_object and source_objects
        self.source_bucket = source_bucket
        self.source_object = source_object
        self.source_objects = source_objects
        self.destination_bucket = destination_bucket
        self.destination_object = destination_object
        self.move_object = move_object
        self.replace = replace
        self.gcp_conn_id = gcp_conn_id
        self.last_modified_time = last_modified_time
        self.maximum_modified_time = maximum_modified_time
        self.is_older_than = is_older_than
        self.impersonation_chain = impersonation_chain
        self.source_object_required = source_object_required
        self.exact_match = exact_match
        self.match_glob = match_glob
        self.delimiter = delimiter
        if self.delimiter:
            raise ValueError("Usage of 'delimiter' is deprecated, please use 'match_glob' instead")

Is this a bad pattern? (In my mind, perhaps the check for wildcard could be after templates are rendered, like in execute)

[potiuk]

Not sure if I understand the problem. Your question does not read it English (at least for me)
But the best way if you miss something is to contribute it back,. I am not even sure what you are trying to do but certainly "skipping" things in your custom constructor is someothing that you should spend time on figuring out rather than maintainers.