Bulk Update or Reassign Schedules

  • 11 March 2021
  • 1 reply
  • 303 views

Hi all,

One common request we get is how to bulk update or reassign schedules. A common use case for this is when users leave the company. First and foremost, do not delete the user before updating the schedules as the schedules and any information on them will be lost.

Generally, we recommend using the API for mass changes in schedules. I’ve written a template below on how to do a bulk re-assign of schedules using the Looker python SDK, but feel free to adjust it in any way to fit your use case. Please read through the commented notes!

import json

import looker_sdk
from looker_sdk import models

# Note that this is assuming we've downloaded the Looker python sdk and that we've set up the ini file in the same folder as this program
# Please refer to https://github.com/looker-open-source/sdk-codegen/tree/main/python for how to download the sdk and create the ini file
sdk = looker_sdk.init31("looker.ini")

# First create a Look that contains a list of scheduled plans & their cron tab that need to be updated from System Activity
# Replace the look_id in the call below
scheduled_plans = sdk.run_look(look_id=11275, result_format='json')
scheduled_plans = json.loads(scheduled_plans)

# Iterates through all the schedules in the list
# replace user_id here with the new user id
for i in scheduled_plans:
json_query = {
"user_id": 1430,
"crontab": i['scheduled_plan.cron_schedule']
}
body = json.dumps(json_query)
sdk.update_scheduled_plan(i['scheduled_plan.id'], body)

Hope this helps out! Any comments or recommendations on making the code better are always welcome!

Cheers,

Annie

Looker DCL 


1 reply

Userlevel 2

Not knowing Annie already wrote this script, I wrote a similar script :sweat_smile: …

I defined a function that would transfer all schedules of user A to user B by using their email addresses as parameters, so we don’t need to run a separate query in system activity. The function call would look like this: 

update_owner('current_owner_email@looker.com', 'new_owner_email@looker.com')


You can run the script on-the-fly with Google Colab: check the link in my repo

If you want to run as a python script, additional code with the sys and os module is needed. 
 

# Install Looker SDK
!pip install looker_sdk
import looker_sdk

# Login
os.environ['LOOKERSDK_BASE_URL'] = 'XXX'
os.environ['LOOKERSDK_CLIENT_ID'] = 'YYY'
os.environ['LOOKERSDK_CLIENT_SECRET'] = 'ZZZ'

#Initialize the SDK
sdk = looker_sdk.init31() # or init40() for v4.0 API


# Return the user id of by looking up email
def find_user_id(email: str):
user_id = sdk.search_users(email=email)
if len(user_id) == 0:
return 'There is no user associated with this email'
else:
return user_id[0]['id']

# Return all schedules of a particular user id
def find_schedules(user_id: int):
result = {}
schedule_plans = sdk.all_scheduled_plans(user_id=user_id)
for i in schedule_plans:
result[i['name']] = i['id']
return result


# Transfer all schedules of a user to a new user.
# If additional condition is needed, please adjust in the loop for sdk.update_schedule_plan()

def update_owner(current_owner_email: str, new_owner_email: str):
current_owner_id = find_user_id(current_owner_email)
new_owner_id = find_user_id(new_owner_email)

if type(new_owner_id) != int and type(new_owner_id) != int:
return "The email addresses for both the current owner and the new owner are not associatd with any Looker user id"

elif type(current_owner_id) != int:
return "The email address for the current owner is not associated with any Looker user id"

elif type(new_owner_id) != int:
return "The email address for the new owner is not associated with any Looker user id"

else:
body = {}
body['user_id'] = new_owner_id
find = find_schedules(current_owner_id)
for i in find.values():
sdk.update_scheduled_plan(i,body)

if find == find_schedules(new_owner_id) and find_schedules(current_owner_id) == {}:

result = "Successfully transfer the following schedules from " + current_owner_email + " to " + new_owner_email
print(result)
for i in find:
print(i)
return None

else:
return "Something goes wrong, please check if email is correct or if the user has any schedules."

 

Reply