Usage
This package only contains two classes you can inherit from BaseAction and AdvancedBaseAction. It simplifies the configuration of a new action by registering with the event hub and converting the events emitted from the server to a more pythonic data representation.
The example actions below can be run directly or added to your ftrack connect plugin path for more information see the main ftrack documentation
Example action
This example registers a new action which only shows up when a single version is selected in the ftrack interface.
# :coding: utf-8
import logging
import ftrack_api
from ftrack_action_handler.action import BaseAction
class MyCustomAction(BaseAction):
'''Custom action.'''
#: A unique identifier for your action.
identifier = 'my.custom.action'
#: A descriptive string identifying your action to the user.
label = 'My Action'
#: Optionally set to differentiate actions with the same label
variant = None
#: A verbose descriptive text for you action
description = 'This is an example action'
def discover(self, session, entities, event):
'''Return True if we can handle the discovery.'''
# TODO: Modify to fit your needs.
# Example, only allow a single asset version as selection.
if len(entities) != 1:
return
entity_type, entity_id = entities[0]
if entity_type != 'AssetVersion':
return
return True
def launch(self, session, entities, event):
'''Callback action'''
for entity_type, entity_id in entities:
entity = session.get(entity_type, entity_id)
# TODO: Do something with the entity.
return True
def register(session, **kw):
'''Register plugin. Called when used as an plugin.'''
# Validate that session is an instance of ftrack_api.Session. If not,
# assume that register is being called from an old or incompatible API and
# return without doing anything.
if not isinstance(session, ftrack_api.session.Session):
return
action_handler = MyCustomAction(session)
action_handler.register()
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
session = ftrack_api.Session()
register(session)
# Wait for events
logging.info('Registered actions and listening for events. Use Ctrl-C to abort.')
session.event_hub.wait()
Example action with interface
Display a interface asking the user for input in order to find and replace words in a specified attribute name on the currently selected items.
# :coding: utf-8
import logging
import ftrack_api
from ftrack_action_handler.action import BaseAction
class FindAndReplace(BaseAction):
'''Example action with interface to find and replace in text attributes.'''
label = 'find and replace'
identifier = 'ftrack.test.find_and_replace'
def discover(self, session, entities, event):
if not self.validate_selection(entities):
return super(FindAndReplace, self).discover(
session, entities, event
)
return True
def launch(self, session, entities, event):
'''Callback method for action.'''
self.logger.info(
u'Launching action with selection {0}'.format(entities)
)
# Validate selection and abort if not valid
if not self.validate_selection(entities):
self.logger.warning(
'Selection is not valid, aborting action'
)
return
values = event['data'].get('values',{})
attribute = values.get('attribute')
find = values.get('find')
replace = values.get('replace')
self.find_and_replace(
session, entities, attribute, find, replace
)
try:
session.commit()
except:
# Commit failed, rollback session and re-raise.
session.rollback()
raise
return {
'success': True,
'message': 'Find and replace "{0}" with "{1}" on attribute "{2}"'.format(
str(find), str(replace), attribute
)
}
def find_and_replace(self, session, entities, attribute, find, replace):
'''Find and replace *find* and *replace* in *attribute* for *selection*.'''
for entity_type, entity_id in entities:
entity = session.get(entity_type, entity_id)
if entity:
value = entity.get(attribute)
if not isinstance(value, basestring):
self.logger.info(
'Ignoring attribute {0!r} with non-string value'.format(attribute)
)
continue
entity.update({
attribute: value.replace(find, replace)
})
def validate_selection(self, entities):
'''Return True if *entities* is valid'''
# Replace with custom logic for validating selection.
# For example check the length or entityType of items in selection.
return True
def interface(self, session, entities, event):
values = event['data'].get('values', {})
if (
not values or not (
values.get('attribute') and
values.get('find') and
values.get('replace')
)
):
# Valid attributes to update.
attributes = [{
'label': 'Name',
'value': 'name'
}, {
'label': 'Description',
'value': 'description'
}]
return [
{
'label': 'Attribute',
'type': 'enumerator',
'name': 'attribute',
'value': attributes[0]['value'],
'data': attributes
}, {
'type': 'text',
'label': 'Find',
'name': 'find'
}, {
'type': 'text',
'label': 'Replace',
'name': 'replace'
}
]
def register(session, **kw):
'''Register plugin. Called when used as an plugin.'''
# Validate that session is an instance of ftrack_api.Session. If not,
# assume that register is being called from an old or incompatible API and
# return without doing anything.
if not isinstance(session, ftrack_api.session.Session):
return
action_handler = FindAndReplace(session)
action_handler.register()
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
session = ftrack_api.Session()
register(session)
# Wait for events
logging.info('Registered actions and listening for events. Use Ctrl-C to abort.')
session.event_hub.wait()
Example Advanced action
A simple action to remove all entries from the joblist of the triggering user.
# :coding: utf-8
import logging
import ftrack_api
from ftrack_action_handler.action import AdvancedBaseAction
class ClearJobList(AdvancedBaseAction):
identifier = "clear_joblist"
label = "Clear Joblist"
icon = "clear_joblist"
action_description = "Removes all entries from the joblist of the triggering user"
allowed_roles = []
allowed_types = []
allow_empty_context = True
def launch(self, session, entities, event):
user = self.get_action_user(event)
success_message = {
"success": True,
"message": u"Joblist cleared"
}
error_message = {
"success": False,
"message": u"An error occured during job list clearing."
}
try:
user_id = user["id"]
job_query = self.session.query(
u"select id, user, user_id"
u" from Job where user_id = ‘{0}’".format(
user_id
)
).all()
for job in job_query:
self.logger.debug(
u"Removing job: {0}".format(job)
)
self.session.delete(job)
self.session.commit()
except Exception as e:
self.logger.error(str(e))
return error_message
return success_message
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO)
session = ftrack_api.Session()
action = ClearJobList(session=session)
action.register(standalone=True)