Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support project_id argument in BigQueryGetDataOperator #25782

Merged
merged 2 commits into from
Aug 20, 2022
Merged

Support project_id argument in BigQueryGetDataOperator #25782

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
71d7d5a
Support project_id argument in BigQueryGetDataOperator
sudohainguyen Aug 18, 2022
3afd9ce
update test case for BigQueryGetDataOperator
sudohainguyen Aug 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions airflow/providers/google/cloud/operators/bigquery.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,13 +361,16 @@ class BigQueryGetDataOperator(BaseOperator):
task_id='get_data_from_bq',
dataset_id='test_dataset',
table_id='Transaction_partitions',
project_id='internal-gcp-project',
max_results=100,
selected_fields='DATE',
gcp_conn_id='airflow-conn-id'
)

:param dataset_id: The dataset ID of the requested table. (templated)
:param table_id: The table ID of the requested table. (templated)
:param project_id: (Optional) The name of the project where the data
will be returned from. (templated)
:param max_results: The maximum number of records (rows) to be fetched
from the table. (templated)
:param selected_fields: List of fields to return (comma-separated). If
Expand All @@ -390,6 +393,7 @@ class BigQueryGetDataOperator(BaseOperator):
template_fields: Sequence[str] = (
'dataset_id',
'table_id',
'project_id',
'max_results',
'selected_fields',
'impersonation_chain',
Expand All @@ -401,6 +405,7 @@ def __init__(
*,
dataset_id: str,
table_id: str,
project_id: Optional[str] = None,
max_results: int = 100,
selected_fields: Optional[str] = None,
gcp_conn_id: str = 'google_cloud_default',
Expand All @@ -419,6 +424,7 @@ def __init__(
self.delegate_to = delegate_to
self.location = location
self.impersonation_chain = impersonation_chain
self.project_id = project_id

def execute(self, context: 'Context') -> list:
self.log.info(
Expand All @@ -445,6 +451,7 @@ def execute(self, context: 'Context') -> list:
max_results=self.max_results,
selected_fields=self.selected_fields,
location=self.location,
project_id=self.project_id,
)

self.log.info('Total extracted rows: %s', len(rows))
Expand Down