Source code for parsons.ngpvan.bulk_import

"""NGPVAN Bulk Import Endpoints"""
from parsons.etl.table import Table
from parsons.utilities import cloud_storage

import logging
import uuid
import csv

logger = logging.getLogger(__name__)

[docs]class BulkImport(object): def __init__(self): pass
[docs] def get_bulk_import_resources(self): """ Get bulk import resources that available to the user. These define the types of bulk imports that you can run. These might include ``Contacts``, ``ActivistCodes``, ``ContactsActivistCodes`` and others. `Returns:` list A list of resources. """ r = self.connection.get_request('bulkImportJobs/resources')'Found {len(r)} bulk import resources.') return r
[docs] def get_bulk_import_job(self, job_id): """ Get a bulk import job status. `Args:` job_id : int The bulk import job id. `Returns:` dict The bulk import job """ r = self.connection.get_request(f'bulkImportJobs/{job_id}')'Found bulk import job {job_id}.') return r
[docs] def get_bulk_import_job_results(self, job_id): """ Get result file of a bulk upload job. This will include one row per record processed as well as the status of each. If the job results have not been generated, either due to an error in the process or the fact the job is still processing, it will return ``None``. `Args:` job_id: int The bulk import job id. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ r = self.get_bulk_import_job(job_id)"Bulk Import Job Status: {r['status']}") if r['status'] == 'Completed': return Table.from_csv(r['resultFiles'][0]['url']) return None
[docs] def get_bulk_import_mapping_types(self): """ Get bulk import mapping types. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ tbl = Table(self.connection.get_request('bulkImportMappingTypes'))'Found {tbl.num_rows} bulk import mapping types.') return tbl
[docs] def get_bulk_import_mapping_type(self, type_name): """ Get a single bulk import mapping type. `Args:` type_name: str `Returns`: dict A mapping type json """ r = self.connection.get_request(f'bulkImportMappingTypes/{type_name}')'Found {type_name} bulk import mapping type.') return r
[docs] def get_bulk_import_mapping_type_fields(self, type_name, field_name): """ Get data about a field in a mapping type. `Args:` type_name: str The mapping type name field_name: str The field name `Returns:` dict A mapping type fields json """ r = self.connection.get_request(f'bulkImportMappingTypes/{type_name}/{field_name}/values')'Found {type_name} bulk import mapping type field values.') return r
def post_bulk_import(self, tbl, url_type, resource_type, mapping_types, description, result_fields=None, **url_kwargs): # Internal method to post bulk imports. # Move to cloud storage file_name = str(uuid.uuid1()) url = cloud_storage.post_file(tbl, url_type, file_path=file_name + '.zip', quoting=csv.QUOTE_ALL, **url_kwargs)'Table uploaded to {url_type}.') # Generate request json json = {"description": description, "file": { "columnDelimiter": 'csv', "columns": [{'name': c} for c in tbl.columns], "fileName": file_name + '.csv', "hasHeader": "True", "hasQuotes": "True", "sourceUrl": url}, "actions": [{"resultFileSizeKbLimit": 5000, "resourceType": resource_type, "actionType": "loadMappedFile", "mappingTypes": mapping_types}] } if result_fields: result_fields = [{'name': c} for c in result_fields] json['actions'][0]['columnsToIncludeInResultsFile'] = result_fields r = self.connection.post_request('bulkImportJobs', json=json)"Bulk upload {r['jobId']} created.") return r['jobId']
[docs] def bulk_apply_activist_codes(self, tbl, url_type, **url_kwargs): """ Bulk apply activist codes. The table may include the following columns. The first column must be ``vanid``. .. list-table:: :widths: 25 25 50 :header-rows: 1 * - Column Name - Required - Description * - ``vanid`` - Yes - A valid VANID primary key * - ``activistcodeid`` - Yes - A valid activist code id * - ``datecanvassed`` - No - An ISO formatted date * - ``contacttypeid`` - No - The method of contact. `Args:` table: Parsons table A Parsons table. url_type: str The cloud file storage to use to post the file (``S3`` or ``GCS``). See :ref:`Cloud Storage <cloud-storage>` for more details. **url_kwargs: kwargs Arguments to configure your cloud storage url type. See :ref:`Cloud Storage <cloud-storage>` for more details. `Returns:` int The bulk import job id """ return self.post_bulk_import(tbl, url_type, 'ContactsActivistCodes', [{"name": "ActivistCode"}], 'Activist Code Upload', **url_kwargs)
[docs] def bulk_upsert_contacts(self, tbl, url_type, result_fields=None, **url_kwargs): """ Bulk create or update contact records. Provide a Parsons table of contact data to create or update records. .. note:: * The first column of the table must be VANID. * The other columns can be a combination of the columns listed below. The valid column names also accept permutations with underscores, spaces and capitalization (e.g. ``phonenumber`` = ``Phone_Number``). **Table Fields** .. list-table:: :widths: 500 100 10 :header-rows: 1 * - Column - Valid Column Names - Notes * - VANID - ``vanid`` - * - Voter VAN ID - ``votervanid`` - The contact's MyVoters VANID * - External ID - ``externalid``, ``id``, ``pk``, ``voterbaseid`` - An external id to be stored. * - **PII** - - * - First Name - ``fn``, ``firstname``, ``last`` - * - Middle Name - ``mn``, ``middlename``, ``middle`` - * - Last Name - ``ln``, ``lastname``, ``last`` - * - Date of Birth - ``dob``, ``dateofbirth`` ``birthdate`` - What type of thing does this need? * - Sex - ``sex``, ``gender`` - * - **Physical Address** - - * - Address Line 1 - ``addressline1``, ``address1``, ``address`` - * - Address Line 2 - ``addressline2``, ``address2`` - * - Address Line 3 - ``addressline3``, ``address3`` - * - City - ``city`` - * - State Or Province - ``state``, ``st``, ``stateorprovince`` - * - Country Code - ``countrycode``, ``country`` - A valid two character country code (e.g. ``US``) * - Display As Entered - ``displayasentered`` - Required values are ``Y`` and ``N``. Determines if the address is processed through address correction. * - **Phones** - - * - Cell Phone - ``cellphone``, ``cell`` - * - Cell Phone Country Code - ``cellcountrycode``, ``cellphonecountrycode`` - A valid two digit country code (e.g. ``01``) * - Home Phone - ``homephone``, ``home``, ``phone`` - * - Home Phone Country Code - ``homecountrycode``, ``homephonecountrycode`` - * - **Email** - - * - Email - ``email``, ``emailaddress`` - `Args:` table: Parsons table A Parsons table. url_type: str The cloud file storage to use to post the file. Currently only ``S3``. results_fields: list A list of fields to include in the results file. **url_kwargs: kwargs Arguments to configure your cloud storage url type. See :ref:`Cloud Storage <cloud-storage>` for more details. `Returns:` int The bulk import job id """ tbl = tbl.map_columns(COLUMN_MAP, exact_match=False) return self.post_bulk_import(tbl, url_type, 'Contacts', [{'name': 'CreateOrUpdateContact'}], 'Create Or Update Contact Records', result_fields=result_fields, **url_kwargs)
def create_mapping_types(self, tbl): # Internal method to generate the correct mapping types based on # the columns passed in the table. Not in use yet. mapping_types = [] # If one of the following columns is found in the table, then add # that mapping type. mp = [('firstname', '') ('Email', 'Email'), ('MailingAddress', 'MailingAddress'), ('Phone', 'Phones'), ('ApplyContactCustomFields', 'CustomFieldGroupId')] for col in tbl.columns: for i in mp: if col.lower() == i[0].lower(): mapping_types.append({'name': i[1]}) return mapping_types
# This is a column mapper that is used to accept additional column names and provide # flexibility for the user. COLUMN_MAP = {'firstname': ['fn', 'first'], 'middlename': ['mn', 'middle'], 'lastname': ['ln', 'last'], 'dob': ['dateofbirth', 'birthdate'], 'sex': ['gender'], 'addressline1': ['address', 'addressline1', 'address1'], 'addressline2': ['addressline2', 'address2'], 'addressline3': ['addressline3', 'address3'], 'city': [], 'stateorprovince': ['state', 'st'], 'countrycode': ['country'], 'displayasentered': [], 'cellphone': ['cell'], 'cellphonecountrycode': ['cellcountrycode'], 'phone': ['home', 'homephone'], 'phonecountrycode': ['phonecountrycode'], 'email': ['emailaddress']}