New Driver Training

Assign onboarding training courses automatically to new drivers added to your Samsara organization.

Overview

This script automates the process of assigning training courses to new drivers in your organization. It interacts with Samsara APIs to fetch drivers based on tags, check existing assignments, and create new assignments for drivers based on their new hire status.

This guide assumes you are already familiar with the basics of the Samsara Training Product and the overall GET Training Assignments, POST Training Assignments, and GET Training Courses APIs.

Example use cases

• An HR manager wants to assign the “New Hire Training” course to new drivers in the organization.
• An HR manager wants to assign an onboarding course to new drivers that are added to Samsara and added to tag “New Driver”.

Trigger conditions

• A new driver is added
• Specify driver tags for assignment

Action

• Assign relevant onboarding training course with defined due dates

Execution flow

1. Generate lookback timestamp

   Generate a timestamp for filtering new drivers based on NEW_DRIVER_LOOKBACK_DAYS.

2. Retrieve new drivers

   Call get_drivers to retrieve new drivers and extract their IDs.

3. Filter already-assigned drivers

   Check if any of the newly created drivers has already been assigned the course and omit those drivers.

4. Generate due date

   Generate a due date timestamp for training completion.

5. Assign training

   Call assign_training to assign the training course to the drivers.

Script

To run certain automations, you will need to set up a development environment. These code snippets can be copied but will need to be customized to match your organization’s specific use case.

from datetime import datetime, timedelta
import requests

# API URLs and token
LIST_DRIVERS_API = 'https://api.samsara.com/fleet/drivers'
ASSIGN_TRAINING_API = 'https://api.samsara.com/training-assignments'
STREAM_TRAINING_ASSIGNMENT_API = 'https://api.samsara.com/training-assignments/stream'
API_TOKEN = 'Insert your API token'
HEADERS = {'Accept': 'application/json', 'Authorization': f'Bearer {API_TOKEN}'}

# Constants
COURSE_ID = 'Insert course ID'
TARGET_DRIVER_TAG_IDS = ['Insert the driver tag ID']
NEW_DRIVERS_LOOKBACK_DAYS = -1
COURSE_COMPLETION_DUE_DAYS = 14
TRAINING_LOOKBACK_DAYS = -1

def get_drivers(tag_ids, created_after_timestamp):
    drivers = []
    after = ''
    has_next_page = True
    params = {'createdAfterTime': created_after_timestamp, 'tagIds': tag_ids}

    while has_next_page:
        params['after'] = after
        response = requests.get(LIST_DRIVERS_API, headers=HEADERS, params=params)
        data = response.json()
        drivers.extend(data.get('data', []))

        if data.get('pagination', {}).get('hasNextPage'):
            after = data['pagination']['endCursor']
        else:
            has_next_page = False

    return drivers

def get_training_assignment_details(assignment_interval_timestamp, course_id):
    training_assignments = []
    after = ''
    params = {'startTime': assignment_interval_timestamp, 'courseIds': [course_id]}
    has_next_page = True
    while has_next_page:
        params['after'] = after
        response = requests.get(STREAM_TRAINING_ASSIGNMENT_API, headers=HEADERS, params=params)
        data = response.json()
        training_assignments.extend(data['data'])
        if data.get('pagination', {}).get('hasNextPage'):
            after = data['pagination']['endCursor']
        else:
            has_next_page = False

    return training_assignments

def assign_training(driver_id_list, course_id, due_date):
    params = {
        'learnerIds': ','.join([f'driver-{id}' for id in driver_id_list]),
        'courseId': course_id,
        'dueAtTime': due_date,
    }
    requests.post(ASSIGN_TRAINING_API, headers=HEADERS, params=params)

def generate_rfc3339_timestamp(delta_days):
    target_date = (datetime.now() + timedelta(days=delta_days)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    return target_date.strftime('%Y-%m-%dT%H:%M:%SZ')

def main():
    driver_lookback_days_rfc3339 = generate_rfc3339_timestamp(NEW_DRIVERS_LOOKBACK_DAYS)
    newly_created_drivers = get_drivers(TARGET_DRIVER_TAG_IDS, driver_lookback_days_rfc3339)
    driver_id_list = [driver['id'] for driver in newly_created_drivers]

    drivers_assigned_training = []
    training_lookback_days_rfc3339 = generate_rfc3339_timestamp(TRAINING_LOOKBACK_DAYS)
    assignment_details = get_training_assignment_details(training_lookback_days_rfc3339, COURSE_ID)

    for assignment in assignment_details:
        if assignment['learner']['type'] == 'driver':
            drivers_assigned_training.append(assignment['learner']['id'])

    drivers_requiring_training = [
        driver for driver in driver_id_list if driver not in drivers_assigned_training
    ]
    drivers_requiring_training = ','.join([f'driver-{id}' for id in drivers_requiring_training])

    course_due_date_rfc3339 = generate_rfc3339_timestamp(COURSE_COMPLETION_DUE_DAYS)
    assign_training(driver_id_list, COURSE_ID, course_due_date_rfc3339)

if __name__ == '__main__':
    main()

Appendix

Constant definitions

Constant	Type	Description
COURSE_ID	String	ID of the training course that will be assigned.
TARGET_DRIVER_TAG_IDS	List of Strings	List of tag IDs used to filter specific drivers.
NEW_DRIVER_LOOKBACK_DAYS	Integer	Timeframe (in days) for fetching new drivers.
COURSE_COMPLETION_DUE_DAYS	Integer	Number of days from today to set as the due date for training assignments.
TRAINING_LOOKBACK_DAYS	Integer	Number of days to look back for training assignments.

Function definitions

Function	Description
get_drivers	Retrieves drivers created after the specified timestamp and matching the provided tag IDs.
get_training_assignment_details	Retrieves a stream of training assignments filtered by assignment_interval_timestamp and course_id.
assign_training	Creates training assignments for drivers.
generate_rfc3339_timestamp	Generates a UTC midnight RFC3339 timestamp by adding or subtracting delta_days. See Timestamps.