Source code for parsons.ngpvan.events

"""NGPVAN Events Endpoints"""

from parsons.etl.table import Table
import logging

logger = logging.getLogger(__name__)


[docs]class Events(object): def __init__(self, van_connection): self.connection = van_connection
[docs] def get_events(self, code_ids=None, event_type_ids=None, rep_event_id=None, starting_after=None, starting_before=None, district_field=None, expand_fields=['locations', 'codes', 'shifts', 'roles', 'notes', 'financialProgram', 'ticketCategories', 'onlineForms']): """ Get events. `Args:` code_ids: str Filter by code id. event_type_ids: str Filter by event_type_ids. rep_event_id: str Filters to recurring events that are recurrences the passed event id. starting_after: str Events beginning after ``iso8601`` formatted date. starting_before: str Events beginning before ``iso8601`` formatted date. district_field: str Filter by district field. expand_fields: list A list of fields for which to include data. If a field is omitted, ``None`` will be returned for that field. Can be ``locations``, ``codes``, ``shifts``,``roles``, ``notes``, ``financialProgram``, ``ticketCategories``, ``onlineForms``. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ if expand_fields: expand_fields = ','.join(expand_fields) params = {'codeIds': code_ids, 'eventTypeIds': event_type_ids, 'inRepetitionWithEventId': rep_event_id, 'startingAfter': starting_after, 'startingBefore': starting_before, 'districtFieldValue': district_field, '$top': 50, '$expand': expand_fields } tbl = Table(self.connection.get_request('events', params=params)) logger.info(f'Found {tbl.num_rows} events.') return tbl
[docs] def get_event(self, event_id, expand_fields=['locations', 'codes', 'shifts', 'roles', 'notes', 'financialProgram', 'ticketCategories', 'voterRegistrationBatches']): """ Get an event. `Args:` event_id: int The event id. expand_fields: list A list of fields for which to include data. If a field is omitted, ``None`` will be returned for that field. Can be ``locations``, ``codes``, ``shifts``, ``roles``, ``notes``, ``financialProgram``, ``ticketCategories``, ``voterRegistrationBatches`.` `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ if expand_fields: expand_fields = ','.join(expand_fields) r = self.connection.get_request(f'events/{event_id}', params={'$expand': expand_fields}) logger.info(f'Found event {event_id}.') return r
[docs] def create_event(self, name, short_name, start_date, end_date, event_type_id, roles, shifts=None, description=None, editable=False, publicly_viewable=False, location_ids=None, code_ids=None, notes=None, district_field_value=None, voter_registration_batches=None): """ Create an event `Args:` name: str A name for this event, no longer than 500 characters. short_name: str A shorter name for this event, no longer than 12 characters. start_date: str The start date and time for this event. end_date: str The end date and time for this event that is after ``start_date`` event_type_id: int A valid event type id. roles: list A list of valid role ids that correspond the with the event type. shifts: A list of dicts with shifts formatted as: .. highlight:: python .. code-block:: python [ { 'name': 'Shift 1', 'start_time': '12-31-2018T12:00:00', 'end_time': '12-31-2018T13:00:00' } { 'name': 'Shift 2', 'start_time': '12-31-2018T13:00:00', 'end_time': '12-31-2018T14:00:00' } ] description: str An optional description for this Event, no longer than 500 characters. editable: boolean If ``True``, prevents modification of this event by any users other than the user associated the API key. Setting this to true effectively makes the event read-only in the VAN interface. publicly_viewable: boolean Used by NGP VAN’s website platform to indicate whether this event can be viewed publicly. location_ids: list A list of location_ids where the event is taking place code_ids: list A list of codes that are applied to this event for organizational purposes. Note that at most one source code and any number of tags, may be applied to an event. notes: list A list of notes `Returns:` int The event code. """ if shifts is None: shifts = [{'name': 'Default Shift', 'startTime': start_date, 'endTime': end_date}] else: shifts = [{'name': s['name'], 'startTime': s['start_time'], 'endTime': s['end_time']} for s in shifts] event = {'name': name, 'shortName': short_name, 'description': description, 'startDate': start_date, 'endDate': end_date, 'eventType': {'eventTypeId': event_type_id}, 'isOnlyEditableByCreatingUser': str(editable).lower(), 'isPubliclyViewable': publicly_viewable, 'notes': notes, 'shifts': shifts, 'roles': [{'roleId': r} for r in roles], 'districtFieldValue': district_field_value, 'voterRegistrationBatches': voter_registration_batches } if location_ids: event['locations'] = [{'locationId': l} for l in location_ids], # noqa E741 if code_ids: event['codes'] = [{'codeID': c} for c in code_ids] r = self.connection.post_request('events', json=event) logger.info(f'Event {r} created.') return r
[docs] def delete_event(self, event_id): """ Delete an event. `Args:` event_id: int The event id. `Returns:` ``None`` """ r = self.connection.delete_request(f'events/{event_id}') logger.info(f'Event {event_id} deleted.') return r
[docs] def add_event_shift(self, event_id, shift_name, start_time, end_time): """ Add shifts to an event `Args:` event_id: int The event id. shift_name: str The name of the shift start_time: str The start time for the shift (``iso8601`` formatted date). end_time: str The end time of the shift (``iso8601`` formatted date). `Returns:` int The shift id. """ shift = {'name': shift_name, 'startTime': start_time, 'endTime': end_time } r = self.connection.post_request(f'events/{event_id}/shifts', json=shift) logger.info(f'Shift {r} added.') return r
[docs] def get_event_types(self): """ Get event types. `Returns:` Parsons Table See :ref:`parsons-table` for output options. """ tbl = Table(self.connection.get_request('events/types')) logger.info(f'Found {tbl.num_rows} events.') return tbl