Repository: apache/airflow
Pull request: Fix: GCS To BigQuery source_object
Fixes https://github.com/apache/airflow/issues/16008 GCS To BigQuery Operator's parameter `source_object` to accept both str and list

<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.

Feel free to ping committers for the review!

In case of existing issue, reference it using one of the following:

closes: #ISSUE
related: #ISSUE

How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).


Problem statement:
GoogleCloudStorageToBigQueryOperator reads string as a list in parameter source_objects
**Apache Airflow version**:1.10.12
**Environment**: google cloud composer

**What happened**:
When using GoogleCloudStorageToBigQueryOperator and providing string as parameter source_objects, the process is iterating on a the string as a valid list.
For example - 
`cloud_storage_to_bigquery = GoogleCloudStorageToBigQueryOperator(
        bucket = 'bucket',
        source_objects = 'abc',
    )`
Will result in looking into the sources: bucket/a, bucket/b, bucket/c.

**What you expected to happen**:
Throw an error on type (string instead of list).

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

`GoogleCloudStorageToBigQueryOperator` is deprecated. Please use `GCSToBigQueryOperator` from providers.
Note that `source_objects` expect `list` not `str`. You should set `source_objects = ['abc']`
@eladkal seems like the behaviour is the same in GCSToBigQueryOperator.
I get that the parameter should be set as `source_objects = ['abc']` but shouldn't I expect type error when setting it to `source_objects = 'abc'`? instead of iterating on the chars of the string and looking for different paths based on it?
PRs are welcome :)
@eladkal Alright. Can this be assigned to me?
You can just open a PR
BTW probably a better solution would be to allow `str or List[str]` rather than raise error.

Commit message: Fix: GCS To BigQuery source_object #16008

Fix GCS To BigQuery source_object to accept both str and list

Patch to review:
diff --git a/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py b/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
index b9251100ed999..b90286fc1d6dd 100644
--- a/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
+++ b/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
@@ -42,9 +42,9 @@ class GCSToBigQueryOperator(BaseOperator):
 
     :param bucket: The bucket to load from. (templated)
     :type bucket: str
-    :param source_objects: List of Google Cloud Storage URIs to load from. (templated)
+    :param source_objects: String or List of Google Cloud Storage URIs to load from. (templated)
         If source_format is 'DATASTORE_BACKUP', the list must only contain a single URI.
-    :type source_objects: list[str]
+    :type source_objects: str, list[str]
     :param destination_project_dataset_table: The dotted
         ``(<project>.|<project>:)<dataset>.<table>`` BigQuery table to load data into.
         If ``<project>`` is not included, project will be the project defined in
@@ -286,7 +286,7 @@ def execute(self, context):
         else:
             schema_fields = self.schema_fields
 
-        source_uris = [f'gs://{self.bucket}/{source_object}' for source_object in self.source_objects]
+        source_uris = [f'gs://{self.bucket}/{source_object}' for source_object in self.source_objects] if type(self.source_objects) is list else [f'gs://{self.bucket}/{self.source_objects}'] 
         conn = bq_hook.get_conn()
         cursor = conn.cursor()
 


Code review comments:
diff_hunk: @@ -286,7 +286,7 @@ def execute(self, context):
         else:
             schema_fields = self.schema_fields
 
-        source_uris = [f'gs://{self.bucket}/{source_object}' for source_object in self.source_objects]
+        source_uris = [f'gs://{self.bucket}/{source_object}' for source_object in self.source_objects] if type(self.source_objects) is list else [f'gs://{self.bucket}/{self.source_objects}'] 
path: airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
text: @user1:
Wouldn't it be better to convert the string to list and then the code will work just fine without a special case for string?

@author:
But if the input is a list, wouldn't it be converted to a 2-dimensional list?
```
self.source_objects = "file1.csv"
source_object = [self.source_objects]
output: ["file1.csv"]

self.source_objects = ["file1.csv", "file2.csv", "file3.csv"]
source_object = [self.source_objects]
output: [["file1.csv", "file2.csv", "file3.csv"]]
```

@user1:
No. I was thinking of converting the type to List (if needed) so the logic part of the code won't need any modification as it works fine with List type. something like:
```
if isinstance(obj, str):
   obj=[obj]
```

@author:
ahh, I was thinking about that solution too. So, it doesn't have to be one line code?