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") #Corresponding to garbled Japanese characters

    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 #Specify schema if necessary

        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))

Recommended Posts