From 27fc640a9eae70813e9f02cda2c27a4509a04591 Mon Sep 17 00:00:00 2001 From: Xianny <8582774+xianny@users.noreply.github.com> Date: Tue, 8 Jan 2019 13:50:51 -0800 Subject: fetch and save copper (#1472) Fetch and save Copper CRM --- packages/pipeline/src/scripts/pull_copper.ts | 129 +++++++++++++++++++++++++++ 1 file changed, 129 insertions(+) create mode 100644 packages/pipeline/src/scripts/pull_copper.ts (limited to 'packages/pipeline/src/scripts/pull_copper.ts') diff --git a/packages/pipeline/src/scripts/pull_copper.ts b/packages/pipeline/src/scripts/pull_copper.ts new file mode 100644 index 000000000..69814f209 --- /dev/null +++ b/packages/pipeline/src/scripts/pull_copper.ts @@ -0,0 +1,129 @@ +// tslint:disable:no-console +import * as R from 'ramda'; +import { Connection, ConnectionOptions, createConnection, Repository } from 'typeorm'; + +import { CopperEndpoint, CopperSearchParams, CopperSource } from '../data_sources/copper'; +import { CopperActivity, CopperActivityType, CopperCustomField, CopperLead, CopperOpportunity } from '../entities'; +import * as ormConfig from '../ormconfig'; +import { + CopperSearchResponse, + parseActivities, + parseActivityTypes, + parseCustomFields, + parseLeads, + parseOpportunities, +} from '../parsers/copper'; +import { handleError } from '../utils'; +const ONE_SECOND = 1000; +const COPPER_RATE_LIMIT = 10; +let connection: Connection; + +(async () => { + connection = await createConnection(ormConfig as ConnectionOptions); + + const accessToken = process.env.COPPER_ACCESS_TOKEN; + const userEmail = process.env.COPPER_USER_EMAIL; + if (accessToken === undefined || userEmail === undefined) { + throw new Error('Missing required env var: COPPER_ACCESS_TOKEN and/or COPPER_USER_EMAIL'); + } + const source = new CopperSource(COPPER_RATE_LIMIT, accessToken, userEmail); + + const fetchPromises = [ + fetchAndSaveLeadsAsync(source), + fetchAndSaveOpportunitiesAsync(source), + fetchAndSaveActivitiesAsync(source), + fetchAndSaveCustomFieldsAsync(source), + fetchAndSaveActivityTypesAsync(source), + ]; + fetchPromises.forEach(async fn => { + await fn; + }); +})().catch(handleError); + +async function fetchAndSaveLeadsAsync(source: CopperSource): Promise { + const repository = connection.getRepository(CopperLead); + const startTime = await getMaxAsync(connection, 'date_modified', 'raw.copper_leads'); + console.log(`Fetching Copper leads starting from ${startTime}...`); + await fetchAndSaveAsync(CopperEndpoint.Leads, source, startTime, {}, parseLeads, repository); +} + +async function fetchAndSaveOpportunitiesAsync(source: CopperSource): Promise { + const repository = connection.getRepository(CopperOpportunity); + const startTime = await getMaxAsync(connection, 'date_modified', 'raw.copper_opportunities'); + console.log(`Fetching Copper opportunities starting from ${startTime}...`); + await fetchAndSaveAsync( + CopperEndpoint.Opportunities, + source, + startTime, + { sort_by: 'name' }, + parseOpportunities, + repository, + ); +} + +async function fetchAndSaveActivitiesAsync(source: CopperSource): Promise { + const repository = connection.getRepository(CopperActivity); + const startTime = await getMaxAsync(connection, 'date_modified', 'raw.copper_activities'); + const searchParams = { + minimum_activity_date: Math.floor(startTime / ONE_SECOND), + }; + console.log(`Fetching Copper activities starting from ${startTime}...`); + await fetchAndSaveAsync(CopperEndpoint.Activities, source, startTime, searchParams, parseActivities, repository); +} + +async function getMaxAsync(conn: Connection, sortColumn: string, tableName: string): Promise { + const queryResult = await conn.query(`SELECT MAX(${sortColumn}) as _max from ${tableName};`); + if (R.isEmpty(queryResult)) { + return 0; + } else { + return queryResult[0]._max; + } +} + +// (Xianny): Copper API doesn't allow queries to filter by date. To ensure that we are filling in ascending chronological +// order and not missing any records, we are scraping all available pages. If Copper data gets larger, +// it would make sense to search for and start filling from the first page that contains a new record. +// This search would increase our network calls and is not efficient to implement with our current small volume +// of Copper records. +async function fetchAndSaveAsync( + endpoint: CopperEndpoint, + source: CopperSource, + startTime: number, + searchParams: CopperSearchParams, + parseFn: (recs: T[]) => E[], + repository: Repository, +): Promise { + let saved = 0; + const numPages = await source.fetchNumberOfPagesAsync(endpoint); + try { + for (let i = numPages; i > 0; i--) { + console.log(`Fetching page ${i}/${numPages} of ${endpoint}...`); + const raw = await source.fetchSearchResultsAsync(endpoint, { + ...searchParams, + page_number: i, + }); + const newRecords = raw.filter(rec => rec.date_modified * ONE_SECOND > startTime); + const parsed = parseFn(newRecords); + await repository.save(parsed); + saved += newRecords.length; + } + } catch (err) { + console.log(`Error fetching ${endpoint}, stopping: ${err.stack}`); + } finally { + console.log(`Saved ${saved} items from ${endpoint}, done.`); + } +} + +async function fetchAndSaveActivityTypesAsync(source: CopperSource): Promise { + console.log(`Fetching Copper activity types...`); + const activityTypes = await source.fetchActivityTypesAsync(); + const repository = connection.getRepository(CopperActivityType); + await repository.save(parseActivityTypes(activityTypes)); +} + +async function fetchAndSaveCustomFieldsAsync(source: CopperSource): Promise { + console.log(`Fetching Copper custom fields...`); + const customFields = await source.fetchCustomFieldsAsync(); + const repository = connection.getRepository(CopperCustomField); + await repository.save(parseCustomFields(customFields)); +} -- cgit v1.2.3