Title: List outdated gear rules and update gear rules for project(s) with latest gear version.
Date: 6 April 2020
Description:
List all project(s) available and gear rules for the project
Topics that will be covered:
# Install specific packages required for this notebook
!pip install flywheel-sdk tqdm pydicom pandas
# Import packages
from getpass import getpass
import os
from pathlib import Path
import re
import json
import copy
import logging
import pydicom
from tqdm.notebook import tqdm
import pandas as pd
import flywheel
from permission import check_user_permission
# Instantiate a logger
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger('root')
Get a API_KEY. More on this in the Flywheel SDK doc here.
API_KEY = getpass('Enter API_KEY here: ')
Instantiate the Flywheel API client
fw = flywheel.Client(API_KEY if 'API_KEY' in locals() else os.environ.get('FW_KEY'))
Show Flywheel logging information
log.info('You are now logged in as %s to %s', fw.get_current_user()['email'], fw.get_config()['site']['api_url'])
# Subset of projects to look at (list of project.label or None (=all project))
PROJECT_LABEL_SUBSET = None
# Subset of gear name to look at (list of gear.name or None (=all gears))
GEAR_NAME_SUBSET = ['metadata-import-dicom']
Before starting off, we want to check your permission on the Flywheel Instance in order to proceed in this notebook.
min_reqs = {
"site": "user",
"group": "ro",
"project": ['session_templates_manage']
}
GROUP_ID = input('Please enter the Group ID that you will be working with: ')
check_user_permission
will return True if both the group and project meet the minimum requirement, else a compatible list will be printed.
check_user_permission(fw, min_reqs, group=GROUP_ID, project=PROJECT_LABEL)
def get_gear_latest_version(fw, gear_id):
"""Return the latest version a gear.
Lookup is done on gear.name
Args:
fw (flywheel.Client): A flywheel client
gear_id (str): An ID of a flywheel.Gear instance
Return:
(flywheel.Gear or None): The latest version of a matching gear or None
"""
curr_gear = fw.get_gear(gear_id)
gears = fw.get_all_gears(all_versions=True, filter=f'gear.name={curr_gear.gear.name}') # sort on version does not seem to work
if gears:
return sorted(gears, key=lambda x: x.gear.version, reverse=True)[0]
else:
return None
def cleanup_rule(rule, project=None):
"""Return a Rule object from input rule dictionary
Clean the input rule dictionary and return a Rule object
Args:
rule (dict): A dictionary containing rule key/value (e.g. extracted from a rule object).
project (flywheel.Project): A flywheel project instance
Return:
(flywheel.Rule): A flywheel gear rule instance
"""
# For each fixed input, fix the project id
if rule.get('fixed_inputs'):
for fi in rule['fixed_inputs']:
if project:
fi['id'] = project.id
fi['type'] = "project"
if not fi['base']:
del fi['base']
if not fi['found']:
del fi['found']
# Fix all and any fields
if rule.get('all'):
for ar in rule['all']:
if not ar['regex']:
ar['regex'] = False
if rule.get('any'):
for ar in rule['any']:
if not ar['regex']:
ar['regex'] = False
if rule.get('_not'):
for ar in rule['_not']:
if not ar['regex']:
ar['regex'] = False
# Formulate the gear_rule
return flywheel.models.rule.Rule(project_id=rule['project_id'],
gear_id=rule['gear_id'],
name=rule['name'],
config=rule['config'],
fixed_inputs=rule['fixed_inputs'],
auto_update=rule['auto_update'],
any=rule['any'],
all=rule['all'],
_not=rule['_not'],
disabled=rule['disabled'])
Create a list of all projects in flywheel or a subset defined in PROJECT_LABEL_SUBSET
projects = []
for p in fw.projects():
if PROJECT_LABEL_SUBSET:
if p.label in PROJECT_LABEL_SUBSET:
projects.append(p)
else:
projects.append(p)
len(projects)
outdated_gears = []
For each project in projects
, get the gear-rules and check whether or not each gear-rule is using the latest version of the gear. If not, add it to the outdated_gears
list.
for project in tqdm(projects):
p_rules = fw.get_project_rules(project.id)
for i, rule in enumerate(p_rules):
rule = rule.to_dict()
try:
current_gear = fw.get_gear(rule['gear_id'])
except flywheel.ApiException:
outdated_gears.append(f'Project: {project.label}, not gear found for rule #{i}')
continue
latest_gear = get_gear_latest_version(fw, rule['gear_id'])
if latest_gear.gear.version != current_gear.gear.version:
outdated_gears.append(f'Project: {project.label}, Gear name: {current_gear.gear.name}, Current: {current_gear.gear.version}, Latest: {latest_gear.gear.version}')
outdated_gears
For each project in projects
, for each gear-rule in project, update the gear-rule with the latest gear version
for project in projects:
p_rules = fw.get_project_rules(project.id)
updated_rules = copy.deepcopy(p_rules)
# updating rules
for rule in updated_rules:
rule = rule.to_dict()
latest_gear = get_gear_latest_version(fw, rule['gear_id'])
if not GEAR_NAME_SUBSET or (GEAR_NAME_SUBSET and latest_gear.gear.name in GEAR_NAME_SUBSET):
if latest_gear:
log.info('Replacing gear_id in gear rule id=%s with gear_id=%s (previously gear_id=%s)', rule['id'], latest_gear.id, rule['gear_id'])
rule['gear_id'] = latest_gear.id
# removing gear rules
log.info('Removing gear rule id=%s in project id=%s', rule['id'], project.id)
fw.remove_project_rule(project.id, rule['id'])
# cleanup rule object
rule = cleanup_rule(rule)
# adding rule
log.info('Adding new gear rule to project id=%s:\n %s ', project.id, rule)
fw.add_project_rule(project.id, rule)
else:
log.info('No gear matching for gear rule id=%s and gear_id=%s. Skipping.', rule['id'], rule['gear_id'])