Accessing AGOL Notebook Logs with ArcGIS API for Python

[DanielMarkbreiter]

Background

I have an ArcGIS Notebook hosted on ArcGIS Online. I can see all of my scheduled Tasks using the AGOL website as well as their "Task details" (see attached screenshot). I'd like to retrieve this information (specifically when a Task was "Created" and its "Status") via the ArcGIS API for Python. 

Troubleshooting

Looking at the ArcGIS API for Python documentation it appears that there is the `arcgis.gis.tasks` module that will list all Tasks with the TaskManager.search() object method. However, the documentation states that this operation is for "Enterprise configuration 10.8.1+".

Question

Is there a similar operation for ArcGIS Online users? Being able to access that information programmatically would be extremely beneficial for the development of Notebooks that require logs. 

[esptjones]

I ran into this today and figured I would answer this post since it has been awhile and no one has answered it. I don't have a perfect solution using the ArcGIS python API, however, you can view tasks and their runs in AGOL by just making a GET request. I will provide an image and a the code I used to get it below. 

 Here is the code I used if you are interested. It was hastily written so don't judge too much 😊

import requests

from arcgis.gis import GIS

from datetime import datetime

import json

def extract_tasks(response😞

    tasks = response.json()['tasks']

    task_list = []

    for task in tasks:

        next_run = task.get('nextStart')

        if next_run:

            next_run = datetime.fromtimestamp(next_run / 1000).strftime('%Y-%m-%d %H:%M:%S')

        task_list.append(

            {

                'task_id': task.get('id'),

                'task_name': task.get('title'),

                'schedule': task.get('cronSchedule'),

                'next_run':next_run,

                'task_state': task.get('taskState'),

                'active': task.get('active')

            }

        )

    return task_list

def extract_task_runs(response😞

    runs = response.json()['runs']

    results = []

    for run in runs:

        t = run['task']

        result = json.loads(run['result'])

        start_time = result.get('startTime')

        end_time = result.get('endTime')

        if start_time:

            start_time = datetime.fromtimestamp(int(start_time) / 1000)

        if end_time:

            end_time = datetime.fromtimestamp(int(end_time) / 1000)

        if start_time and end_time:

            duration = end_time - start_time

        else:

            duration = None

        results.append(

            {

                'task_id':t.get('id'),

                'task_name': t.get('title'),

                'run_id': run.get('runId'),

                'status': run.get('status'),

                'start_time': start_time.strftime('%Y-%m-%d %H:%M:%S'),

                'end_time': end_time.strftime('%Y-%m-%d %H:%M:%S'),

                'duration': duration,

                'results': result

            }

        )

    results[0]

    return results

# Get the GIS Object

gis = GIS(settings['agol'], settings['username'], settings['password'])

# Extract Token for requests

token = gis._con.token

# Get the username and url key

user = gis._username

url_key = gis.properties.urlKey

# Get a list of all tasks from the organization

tasks_url = f"https://{url_key}.maps.arcgis.com/sharing/rest/community/users/{user}/tasks?f=json&token={token}"

response = requests.get(tasks_url)

if response.status_code != 200:

    raise Exception(f"Failed to get tasks. Status code: {response.status_code}")

task_list = extract_tasks(response)

print(task_list[0])

# Get the runs for the first task

task_id = task_list[0]['task_id']

url = f"https://{url_key}.maps.arcgis.com/sharing/rest/community/users/{user}/tasks/{task_id}/runs?f=json&token={token}"

response = requests.get(url)

if response.status_code != 200:

    raise Exception(f"Failed to get task runs. Status code: {response.status_code}")

task_runs = extract_task_runs(response)

# Print the first run for demonstration

task_runs[0]