StreamSets DataOps Platform SDK script to update object owners

dataops_change_object_owners.py

#!/usr/bin/env python

'''This script changes ownership of objects from an 'old' user to a 'new' user 
in StreamSets DataOps Platform

Set DRY_RUN to True to generate a list of objects owned by the old user without making any changes.

Set DRY_RUN to False to actually change the ownership of objects from the 'old' to the 'new' user

Objects include:

 - Environments
 - Deployments
 - Engines
 - Connections
 - Fragments
 - Pipelines
 - Jobs
 - Topologies
 - Scheduled Tasks
 - Subscriptions
 
 The current version of this script does not update owners of:
 - Legacy Kubernetes Provisioning Agents
 - Legacy Kubernetes Deployments
 
 
Prerequisites:
 - Python 3.6+; Python 3.9+ preferred
 
 - StreamSets DataOps Platform SDK for Python v5.1+
   See: https://docs.streamsets.com/platform-sdk/latest/learn/installation.html
   
 - DataOps Platform API Credentials for a user with Organization Administrator role
 
'''

# Imports
import sys
from streamsets.sdk import ControlHub


## USER VARIABLES ##############################

# Set DRY_RUN to True to generate a list of objects owned by the old user without making any changes
# Set DRY_RUN to False to actually change the ownership of objects from the'old' to the 'new' user
DRY_RUN = True

# CRED_ID -- Your API Credential CRED_ID.  
CRED_ID = ''

# CRED_TOKEN -- Your API Credential CRED_TOKEN
CRED_TOKEN = ''

# Old User email
OLD_USER_EMAIL = '[email protected]'

# New User email
NEW_USER_EMAIL = '[email protected]'

##############################################

# Prints a divider line to the console
def print_divider_line():
    print('-----------------------------------------------------------')
    
# Method to change object owner
def change_resource_owner(object, object_type, object_identifier, new_user):
    print('Changing ownership of ' + object_type + ' \'' + object_identifier + '\' to ' + new_user.email_address)
    object.acl.resource_owner = new_user.id
    
# Method to get old object's owner permissions (we'll set the same permissions for the new object's owner)
def get_existing_permissions(object, object_type, object_identifier, old_user):
    try:
        return object.acl.permissions.get(subject_id=old_user.id).actions
    except:
        print('Warning: No previous permissions found for ' + object_type + ' \'' + object_identifier + '\' for ' + old_user.email_address + '\n')
        return None
  
# Method to set object owner permissions
def set_permissions(object, object_type, object_identifier, new_user, permissions):
    print('Granting new owner permissions: ' + str(permissions) + '\n')
    permissions = object.acl.permission_builder.build(subject_id=new_user.id, subject_type='USER', actions=permissions)
    object.acl.add_permission(permissions)

# Method to process each object
def handle_object(object, object_type, object_identifier, old_user, new_user):

    # If object is owned by the old user
    if object.acl.resource_owner == old_user.id:
    
        # Get the old owner's permissions
        permissions = get_existing_permissions(object, object_type, object_identifier, old_user)     
        
        # Print the object name and owner's permissions
        print(object_type + ' \'' + object_identifier + '\' with permissions ' + str(permissions)) 
        
        if not DRY_RUN:  
        
            # Change the object's owner
            change_resource_owner(object, object_type, object_identifier, new_user)
            
            # Set the new owner's permissions if the old owner had them
            if permissions is not None:
                set_permissions(object, object_type, object_identifier, new_user, permissions)
    
print_divider_line()
if DRY_RUN:
    print('Script is running in DRY_RUN mode; no changes will be made.')
else:
    print('Script is running in ACTIVE mode; object ownership changes will be made.')
print_divider_line()
    
## Init connection to Control Hub 
sch = ControlHub(
    credential_id=CRED_ID, 
    token=CRED_TOKEN)


## Get the 'old' user
try:
    old_user = sch.users.get(email_address = OLD_USER_EMAIL)
except:
    sys.exit('Error: Could not find user with email ' + OLD_USER_EMAIL)   
    
print('Old user: ' + OLD_USER_EMAIL)

## Get the 'new' user
try:
    new_user = sch.users.get(email_address = NEW_USER_EMAIL)
except:
    sys.exit('Error: Could not find user with email ' + NEW_USER_EMAIL)  
    
print('New user: ' + NEW_USER_EMAIL)
print_divider_line()
print('Objects listed below are owned by ' + old_user.email_address)
if not DRY_RUN:
    print('Objects listed below will have ownership changed to ' + new_user.email_address)
print_divider_line()

## Handle Environments
print('\nEnvironments')
print_divider_line()
for environment in sch.environments:
    handle_object(environment, 'Environment', environment.environment_name, old_user, new_user)

## Deployments
print('\nDeployments')
print_divider_line()
for deployment in sch.deployments:
    handle_object(deployment, 'Deployment', deployment.deployment_name, old_user, new_user)

## Engines
print('\nEngines')
print_divider_line()
for engine in sch.engines:        
    handle_object(engine, 'Engine', engine.engine_url, old_user, new_user)

## Connections
print('\nConnections')
print_divider_line()
for connection in sch.connections:
    handle_object(connection, 'Connection', connection.name, old_user, new_user)

## Fragments
print('\nFragments')
print_divider_line()
for fragment in sch.pipelines.get_all(fragment=True):
    handle_object(fragment, 'Fragment', fragment.name, old_user, new_user)
    
## Pipelines
print('\nPipelines')
print_divider_line()
for pipeline in sch.pipelines:
    handle_object(pipeline, 'Pipeline', pipeline.name, old_user, new_user)

## Jobs
print('\nJobs')
print_divider_line()
for job in sch.jobs:
    handle_object(job, 'Job', job.job_name, old_user, new_user)

## Topologies
print('\nTopologies')
print_divider_line()
for topology in sch.topologies:
    handle_object(topology, 'Topology', topology.topology_name, old_user, new_user)

## Scheduled Tasks
print('\nScheduled Tasks')
print_divider_line()
for task in sch.scheduled_tasks:
    handle_object(task, 'Scheduled Task', task.name, old_user, new_user)
    
## Subscriptions
print('\nSubscriptions')
print_divider_line()
for subscription in sch.subscriptions:
    handle_object(subscription, 'Subscription', subscription.name, old_user, new_user)
    
print('\nDone')