Holen Sie sich LEAD-Daten mit der REST-API von Marketo in Python

Hintergrund

Ich möchte Marketo-Daten erhalten. Notieren Sie sich das Beispiel, um die Leads-Daten in großen Mengen zu erhalten. Es kann auch für andere Datenressourcen verwendet werden.

Bevor du anfängst

Machen Sie sich ein Bild von der REST-API, indem Sie Erste Schritte ausführen REST API

Sie können die Atmosphäre auf der Erklärungsseite für die Massenerfassung erfassen Bulk Extract

Beispielcode für die Massenerfassung

import requests
import json
import time
import os
import pendulum
from google.cloud import bigquery

#↓ Kann über den Verwaltungsbildschirm abgerufen werden
# https://developers.marketo.com/rest-api/
ENDPOINT = 'YOUR_ENDPOINT'
CLIENT_ID = 'YOUR_CLIENT_ID'
CLIENT_SECRET = 'YOUR_CLIENT_SECRET'


class MarketoAPI:
    def __init__(self, target_date):
        self.target_date = target_date
        self.token = self.get_token()
        self.header = {
            'Authorization': 'Bearer {}'.format(self.token),
            'content-type': 'application/json'
        }
        self.export_id = None  #Wird beim Erstellen eines Jobs festgelegt

    # CLIENT_ID und KUNDE_Stellen Sie ein Token von SECRET aus
    #Beachten Sie, dass das vom Verwaltungsbildschirm ausgegebene Token sofort abgelaufen ist.
    def get_token(self):
        URL = '{}/identity/oauth/token?client_id={}&client_secret={}&grant_type=client_credentials'.format(
            ENDPOINT, CLIENT_ID, CLIENT_SECRET
        )
        headers = {'content-type': 'application/json'}
        r = requests.get(URL, headers=headers)

        if r.status_code != requests.codes.ok:
            r.raise_for_status()

        return r.json()['access_token']

    #Dieses Mal werden wir die Bulk Extract API verwenden
    #Erstellen Sie zunächst einen Job, um LEAD zu erhalten
    # https://developers.marketo.com/rest-api/bulk-extract/bulk-lead-extract/
    def create(self):
        URL = '{}/bulk/v1/leads/export/create.json'.format(ENDPOINT)

        payload = {
            #Geben Sie das Feld an, das Sie erhalten möchten
            #Die Felder, die angegeben werden können, sind`GET /rest/v1/leads/describe.json`Kann mit der API von bestätigt werden
            "fields": [
                "id",
                "email",
                "createdAt",
                "updatedAt"
            ],
            "format": "CSV",
            #Geben Sie einen Filter an
            #Der Fileter, der angegeben werden kann, ist in ↓ doc geschrieben
            # https://developers.marketo.com/rest-api/bulk-extract/bulk-lead-extract/
            "filter": {
                "createdAt": {
                    "startAt": self.target_date.isoformat(),
                    "endAt": self.target_date.end_of('day').isoformat()
                }
            }
        }

        r = requests.post(URL, data=json.dumps(payload), headers=self.header)

        if r.status_code != requests.codes.ok:
            r.raise_for_status()

        export_id = r.json()['result'][0]['exportId']
        self.export_id = export_id
        return export_id

    # create()Jobs werden nicht nur durch Erstellen in verarbeitet
    #Sie können auf die Ausführung warten, indem Sie den Job hier in die Warteschlange stellen
    def enqueue(self):
        URL = '{}/bulk/v1/leads/export/{}/enqueue.json'.format(
            ENDPOINT, self.export_id
        )
        r = requests.post(URL, headers=self.header)

        if r.status_code != requests.codes.ok:
            r.raise_for_status()

    #Eine Methode zum Abrufen des Statusp, bis die Verarbeitung des in die Warteschlange gestellten Jobs abgeschlossen ist
    def check_until_done(self):
        URL = '{}/bulk/v1/leads/export/{}/status.json'.format(
            ENDPOINT, self.export_id
        )

        #Warten Sie 3 Minuten vor der Anfrage
        #3 Mal ausführen, bis abgeschlossen
        for i in range(3):
            time.sleep(60 * 3)
            r = requests.get(URL, headers=self.header)
            if r.status_code == requests.codes.ok:
                if r.json()['result'][0]['status'] == 'Completed':
                    return True

        if r.status_code != requests.codes.ok:
            r.raise_for_status()
        else:
            raise('job status is: ' + r.json()['result'][0]['status'])

    #Methode zum Herunterladen der Datei nach Abschluss des Auftrags
    def dl_file(self):
        URL = '{}/bulk/v1/leads/export/{}/file.json'.format(
            ENDPOINT, self.export_id
        )
        r = requests.get(URL, headers=self.header)

        if r.status_code != requests.codes.ok:
            r.raise_for_status()

        with open('./marketo_lead_file_{}.csv'.format(self.target_date.strftime('%Y%m%d')), mode='w') as f:
            f.write(r.text)

Unter der Bedingung dieses Erstellens ist die folgende Datei DL

id,email,createdAt,updatedAt
111111,[email protected],2020-01-02T15:35:37Z,2020-01-02T15:35:36Z
222222,[email protected],2030-01-02T22:58:07Z,2020-01-02T22:58:07Z

• Beachten Sie, dass die Bulk-API keine Daten mit bis zu 500 MB pro Tag laden kann.