Rufen Sie Daten über die Salesforce-API (Bulk-API) in Python ab und laden Sie sie in BigQuery

Hintergrund

Ich möchte mit Python Daten von Salesforce abrufen. Dieses Mal habe ich eine Bibliothek namens simple_salesforce und Bulk-API der Salsesorce-API verwendet. -jp.api_asynch.meta / api_asynch / asynch_api_intro.htm), um die Daten abzurufen und in BigQuery zu laden.

Rund um die Authentifizierung

Bereiten Sie die folgenden drei vor

• username
• password
• security_token --security_token sollte von [Sicherheitstoken zurücksetzen] auf dem Verwaltungsbildschirm ausgegeben werden.
• Referenz: [Schritt 3: Überprüfen des Salesforce-Sicherheitstokens](https://support.miitel.jp/hc/ja/articles/360030487511-Step-3-Salesforce-%E3%81%AE%E3%82 % BB% E3% 82% AD% E3% 83% A5% E3% 83% AA% E3% 83% 86% E3% 82% A3% E3% 83% 88% E3% 83% BC% E3% 82% AF % E3% 83% B3% E3% 81% AE% E7% A2% BA% E8% AA% 8D% E6% 96% B9% E6% B3% 95)

Implementierungsbeispiel

from google.cloud import bigquery
from simple_salesforce import Salesforce
import json
import os


class SalesforceAPI:
    def __init__(self, job_type):
        self.sf = Salesforce(
            username='USERNAME',
            password='PASSWORD',
            security_token='SECURITY_TOKEN'
        )

    def execute(self):
        self.dl_file()
        self.load_to_bq()

    def dl_file(self):
        res = self.sf.bulk.TABLE_NAME.query('SELECT column1, column2 FROM TABLE_NAME')
        with open('dl_file_name', mode='w') as f:
            for d in res:
                f.write(json.dumps(d, ensure_ascii=False) + "\n") #Entspricht verstümmelten japanischen Schriftzeichen

    def load_to_bq(self):
        client = bigquery.Client('project')
        filename = 'file_name'
        dataset_id = 'dataset'
        dataset_ref = client.dataset(dataset_id)
        table_id = 'table_name'
        table_ref = dataset_ref.table(table_id)

        job_config = bigquery.LoadJobConfig()
        job_config.source_format = bigquery.SourceFormat.NEWLINE_DELIMITED_JSON
        job_config.autodetect = True #Geben Sie gegebenenfalls das Schema an

        with open(filename, "rb") as source_file:
            job = client.load_table_from_file(
                source_file, table_ref, job_config=job_config
            )
            job.result()
            print("Loaded {} rows into {}:{}.".format(
                job.output_rows, dataset_id, table_id))

Referenz

Ich habe das stärkste SOQL-Ausführungstool erstellt, an das ich gedacht habe

• Das hier vorgestellte SOQL-Ausführungstool ist praktisch
• Verwenden Sie diese Option, um SOQL für DL zu erstellen