Je veux obtenir des données Marketo. Notez l'exemple pour obtenir les données Leads en masse. Il peut également être utilisé pour d'autres ressources de données.
Obtenez une idée de l'API REST en faisant la mise en route REST API
Vous pouvez saisir l'atmosphère en faisant la page d'explication de l'acquisition en masse Bulk Extract
import requests
import json
import time
import os
import pendulum
from google.cloud import bigquery
#↓ Peut être obtenu à partir de l'écran de gestion
# 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 #Définir lors de la création d'un travail
# CLIENT_ID et CLIENT_Émettre un jeton de SECRET
#Notez que le token émis depuis l'écran de gestion expirera immédiatement.
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']
#Cette fois, nous utiliserons l'API Bulk Extract
#Tout d'abord, créez un emploi pour obtenir du plomb
# https://developers.marketo.com/rest-api/bulk-extract/bulk-lead-extract/
def create(self):
URL = '{}/bulk/v1/leads/export/create.json'.format(ENDPOINT)
payload = {
#Spécifiez le champ que vous souhaitez obtenir
#Les champs qui peuvent être spécifiés sont`GET /rest/v1/leads/describe.json`Peut être confirmé avec l'API de
"fields": [
"id",
"email",
"createdAt",
"updatedAt"
],
"format": "CSV",
#Spécifiez un filtre
#Le fileter qui peut être spécifié est écrit dans ↓ doc
# 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()Les travaux ne sont pas traités simplement en créant dans
#Vous pouvez attendre l'exécution en mettant le travail en file d'attente ici
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()
#Une méthode pour interroger l'étatp jusqu'à ce que le traitement du travail mis en file d'attente soit terminé
def check_until_done(self):
URL = '{}/bulk/v1/leads/export/{}/status.json'.format(
ENDPOINT, self.export_id
)
#Attendez 3 minutes avant la demande
#Exécutez 3 fois jusqu'à la fin
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'])
#Méthode de téléchargement du fichier lorsque le travail est terminé
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)
Sous la condition de cette création, le fichier suivant sera 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
Recommended Posts