Create salesforce to gcs transfer #10760

chipmyersjr · 2020-09-05T20:02:38Z

Add Salesforce to GCS transfer operator. Closes issue #8896

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

ephraimbuddy · 2020-09-05T21:41:15Z

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

+    """
+
+    def __init__(
+        self,


Suggested change

self,

self, *,

nitpick here. Operator arguments are now keyword-only

Thank you. Added this arg.

ephraimbuddy · 2020-09-05T21:44:16Z

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

+    :param query: The query to make to Salesforce.
+        :type query: str


Suggested change

:param query: The query to make to Salesforce.

:type query: str

:param query: The query to make to Salesforce.

:type query: str

I'm thinking that this would give error if not on the same line with the param?

Thanks, fixed formatting for all parameters in this doc string.

ephraimbuddy · 2020-09-05T21:54:07Z

airflow/providers/google/cloud/example_dags/example_salesforce_to_gcs.py

+from airflow.providers.google.cloud.transfers.salesforce_to_gcs import SalesforceToGcsOperator
+from airflow.utils.dates import days_ago
+
+os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "/root/keyfile.json"


Nice work! 👍
System test would make this easy with @pytest.mark.credential_file and I think it(system test) is required for google provider packages. Take a look

airflow/tests/providers/google/cloud/operators/test_bigquery_system.py

Line 28 in ba36f87

@pytest.mark.credential_file(GCP_BIGQUERY_KEY)

Thanks for the example! I am looking into implementing system test for this operator.

eladkal

General comments:

Salesforce hook already support multiple formats (json, csv , ndjson) why do you limit this operator only to csv? It also has write_object_to_file method which seems like you implemented in another way in here.
I wonder how close can we get to the same behavior and parameters as BaseSQLToGCSOperator? (It has relevant params like approx_max_file_size_bytes )

eladkal · 2020-09-06T11:59:29Z

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

+    :type object_name: str
+    :param salesforce_conn_id: the name of the connection that has the parameters
+        we need to connect to Salesforce. The connection should be type `http` and
+        include a user's security token in the `Extras` field.


No need to specify this. It's part of the doc of salesforce connection #10482.
This can change when Airflow will support other connection options for salesforce.

chipmyersjr · 2020-09-06T22:32:45Z

@eladkal Thanks for the feedback. I think using write_object_to_file method is a good call out. Will make this simpler and more flexible so I'm going to try to implement it that way.

In terms of adding other parameters, do you think that adding parameters related to file creation like approx_max_file_size_bytes and field_delimiter to happen within write_object_to_file would make sense? Alternatively the execute method would chunk the results and call write_object_to_file multiple times.

eladkal · 2020-09-07T09:00:29Z

In terms of adding other parameters, do you think that adding parameters related to file creation like approx_max_file_size_bytes and field_delimiter to happen within write_object_to_file would make sense? Alternatively the execute method would chunk the results and call write_object_to_file multiple times.

Ignore my previous comment
I would say that providing the same GCP parameters like FacebookAdsReportToGcsOperator is OK though I'm not sure why it doesn't have delegate_to. Could be that at some point there will be a BaseFileToGcsOperator which will prevent the need from adding delegate_to & impersonation_chain to every operator.
I think @mik-laj or @turbaszek are better to answer what parameters are mandatory for GCP.

I'll review the salesforce part later this week.

turbaszek · 2020-09-13T19:52:40Z

tests/providers/google/cloud/transfers/test_salesforce_to_gcs_system.py

+
+
+@contextmanager
+def provide_facebook_connection(key_file_path: str):


Would you mind moving this function to tests.test_utils.facebook_system_helpers?

Sure. Moved to test_utils. Also this should be named salesforce so renamed as well.

docs/howto/operator/google/transfer/salesforce_to_gcs.rst

turbaszek · 2020-09-13T19:55:36Z

tests/providers/google/cloud/transfers/test_salesforce_to_gcs_system.py

+@pytest.mark.backend("mysql", "postgres")
+@pytest.mark.credential_file(GCP_BIGQUERY_KEY)
+@pytest.mark.credential_file(SALESFORCE_KEY)
+@pytest.mark.system("google.cloud")


Should we also mark it as salesforce?

Sure. Marked as salesforce system test.

turbaszek · 2020-09-14T06:50:53Z

@chipmyersjr can you please take a look at the CI issues? Pylint seems to be sad :<

chipmyersjr · 2020-09-14T07:45:56Z

@turbaszek seems pylint was complaining about some unrelated files. I just did rebase and checks seem to be in in good shape now.

eladkal

Salesforce logic looks good
Can you please add support for template_fields, template_ext (reading query from file) ?

chipmyersjr · 2020-09-19T00:26:18Z

@eladkal thanks for checking. I added templating support for parameters where I felt is was appropriate.

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

turbaszek · 2020-09-27T17:07:53Z

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py

+                filename=path,
+                gzip=self.gzip,
+            )
+            self.log.info("%s uploaded to GCS", path)


It would be nice to return the path to file so it can be easily referenced in downstream operators

hi @turbaszek , 'path' variable here represents the temporary file that gets created
just to be uploaded to GCS and wouldn't be available outside of this context. I'm wondering if it would be more appropriate to pass
this or alternatively the path to the actual GCS object? Also wondering about which would be better to log actually.

Sorry for not being precise, by returning path I thought about returning GCS uri

Ok. I implemented uri return value for execute method.

eladkal · 2020-10-27T19:30:58Z

@turbaszek is something missing to get it merged?

turbaszek · 2020-10-28T07:50:45Z

@chipmyersjr can you please rebase and solve the conflict?

…tting

Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

chipmyersjr · 2020-10-29T04:08:15Z

@turbaszek Should be good to go now.

Adds SalesforceToGcsOperator that allows users to transfer data from Salesforce to GCS bucket. Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

boring-cyborg bot added area:dev-tools area:docs provider:google Google (including GCP) related issues labels Sep 5, 2020

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from 6028042 to 1972774 Compare September 5, 2020 21:06

ephraimbuddy reviewed Sep 5, 2020

View reviewed changes

eladkal reviewed Sep 6, 2020

View reviewed changes

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from 923661b to d90fed6 Compare September 7, 2020 05:15

turbaszek reviewed Sep 13, 2020

View reviewed changes

docs/howto/operator/google/transfer/salesforce_to_gcs.rst Outdated Show resolved Hide resolved

turbaszek reviewed Sep 13, 2020

View reviewed changes

chipmyersjr requested a review from turbaszek September 14, 2020 05:42

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from 0f32335 to bf750e3 Compare September 14, 2020 06:12

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from bf750e3 to 03190e9 Compare September 14, 2020 06:56

turbaszek requested a review from mik-laj September 14, 2020 08:22

turbaszek approved these changes Sep 14, 2020

View reviewed changes

eladkal approved these changes Sep 17, 2020

View reviewed changes

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from 352b4d5 to ee6d552 Compare September 18, 2020 23:24

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from ee6d552 to 17799ca Compare September 19, 2020 18:04

turbaszek added area:providers and removed area:dev-tools labels Sep 27, 2020

turbaszek reviewed Sep 27, 2020

View reviewed changes

airflow/providers/google/cloud/transfers/salesforce_to_gcs.py Show resolved Hide resolved

turbaszek reviewed Sep 27, 2020

View reviewed changes

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from 10dabc5 to fd90bca Compare September 30, 2020 06:00

turbaszek approved these changes Sep 30, 2020

View reviewed changes

ryw removed the area:docs label Oct 27, 2020

turbaszek approved these changes Oct 28, 2020

View reviewed changes

chipmyersjr and others added 10 commits October 28, 2020 19:09

Create salesforce to gcs transfer

88968ec

Add * as arg to SalesforceToGcsOperator init and fix doc string forma…

7b7ad77

…tting

Remove some duplicate documentation regarding salesforce connection

cea18bf

Refactor to use write_object_to_file of salesforce hook

b0fa05d

Add system test for salesforce_to_gcs operator

04671ac

Update docs/howto/operator/google/transfer/salesforce_to_gcs.rst

34baf53

Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

Mark TestSalesforceIntoGCSExample as salesforce system test

b338fbc

Refactor provide_salesforce_connection to test_utils

3076c01

Add templating support to SalesforceToGcsOperator

39ff9f1

SalesforceToGcsOperator.execute returns GCS URI

224f684

chipmyersjr force-pushed the salesforce-to-gcs-transfer branch from fd90bca to 224f684 Compare October 29, 2020 02:10

turbaszek merged commit 2f703df into apache:master Oct 30, 2020

szn pushed a commit to szn/airflow that referenced this pull request Nov 1, 2020

Add SalesforceToGcsOperator (apache#10760)

963c3be

Adds SalesforceToGcsOperator that allows users to transfer data from Salesforce to GCS bucket. Co-authored-by: Tomek Urbaszek <turbaszek@gmail.com>

eladkal mentioned this pull request Nov 4, 2020

Add SalesforceToGcs #8896

Closed

potiuk mentioned this pull request Nov 10, 2020

GCSToBigQueryOperator - Not generating the Unique BQ Job Name #11660

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create salesforce to gcs transfer #10760

Create salesforce to gcs transfer #10760

chipmyersjr commented Sep 5, 2020

ephraimbuddy Sep 5, 2020

chipmyersjr Sep 5, 2020

ephraimbuddy Sep 5, 2020

chipmyersjr Sep 5, 2020

ephraimbuddy Sep 5, 2020 •

edited

Loading

chipmyersjr Sep 5, 2020

eladkal left a comment

eladkal Sep 6, 2020

chipmyersjr commented Sep 6, 2020

eladkal commented Sep 7, 2020

turbaszek Sep 13, 2020

chipmyersjr Sep 14, 2020

turbaszek Sep 13, 2020

chipmyersjr Sep 14, 2020

turbaszek commented Sep 14, 2020

chipmyersjr commented Sep 14, 2020

eladkal left a comment

chipmyersjr commented Sep 19, 2020

turbaszek Sep 27, 2020

chipmyersjr Sep 29, 2020

turbaszek Sep 29, 2020

chipmyersjr Sep 30, 2020

eladkal commented Oct 27, 2020

turbaszek commented Oct 28, 2020

chipmyersjr commented Oct 29, 2020 •

edited

Loading

		:param query: The query to make to Salesforce.
		:type query: str



		@contextmanager
		def provide_facebook_connection(key_file_path: str):

Create salesforce to gcs transfer #10760

Create salesforce to gcs transfer #10760

Conversation

chipmyersjr commented Sep 5, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ephraimbuddy Sep 5, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eladkal left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chipmyersjr commented Sep 6, 2020

eladkal commented Sep 7, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

turbaszek commented Sep 14, 2020

chipmyersjr commented Sep 14, 2020

eladkal left a comment

Choose a reason for hiding this comment

chipmyersjr commented Sep 19, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eladkal commented Oct 27, 2020

turbaszek commented Oct 28, 2020

chipmyersjr commented Oct 29, 2020 • edited Loading

ephraimbuddy Sep 5, 2020 •

edited

Loading

chipmyersjr commented Oct 29, 2020 •

edited

Loading