Using

This package only contains one class BaseAction. 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)