Source code for ftrack_action_handler.action.base

# :coding: utf-8
# :copyright: Copyright (c) 2017-2021 ftrack

import json
import logging
import os
import uuid

logging.basicConfig(level=logging.INFO)

# --------------------------------------------------------------
# Base Action Class.
# --------------------------------------------------------------


[docs]class BaseAction(object): '''Custom Action base class `label` a descriptive string identifing your action. `variant` To group actions together, give them the same label and specify a unique variant per action. `identifier` a unique identifier for your action. `description` a verbose descriptive text for you action ''' label = None variant = None identifier = None description = None icon = None
[docs] def __init__(self, session): '''Expects a ftrack_api.Session instance''' self.logger = logging.getLogger( '{0}.{1}'.format(__name__, self.__class__.__name__) ) if self.label is None: raise ValueError( 'Action missing label.' ) elif self.identifier is None: raise ValueError( 'Action missing identifier.' ) self._session = session
@property def session(self): '''Return current session.''' return self._session
[docs] def register(self, standalone=False): '''Registers the action, subscribing the the discover and launch topics. *standalone* lets the action run in self.session useful for testing and development ''' self.session.event_hub.subscribe( 'topic=ftrack.action.discover', self._discover ) self.session.event_hub.subscribe( 'topic=ftrack.action.launch and data.actionIdentifier={0}'.format( self.identifier ), self._launch ) if standalone: self.logger.debug( 'Action: {0} running as standalone.'.format( self.label ) ) self.session.event_hub.wait()
def _discover(self, event): args = self._translate_event( self.session, event ) accepts = self.discover( self.session, *args ) if accepts: return { 'items': [{ 'icon': self.icon, 'label': self.label, 'variant': self.variant, 'description': self.description, 'actionIdentifier': self.identifier, }] }
[docs] def discover(self, session, entities, event): '''Return true if we can handle the selected entities. *session* is a `ftrack_api.Session` instance *entities* is a list of tuples each containing the entity type and the entity id. If the entity is a hierarchical you will always get the entity type TypedContext, once retrieved through a get operation you will have the "real" entity type ie. example Shot, Sequence or Asset Build. *event* the unmodified original event ''' return False
def _translate_event(self, session, event): '''Return *event* translated structure to be used with the API.''' _selection = event['data'].get('selection', []) _entities = list() for entity in _selection: _entities.append( ( self._get_entity_type(entity), entity.get('entityId') ) ) return [ _entities, event ] def _get_entity_type(self, entity): '''Return translated entity type tht can be used with API.''' # Get entity type and make sure it is lower cased. Most places except # the component tab in the Sidebar will use lower case notation. entity_type = entity.get('entityType').replace('_', '').lower() for schema in self.session.schemas: alias_for = schema.get('alias_for') if ( alias_for and isinstance(alias_for, str) and alias_for.lower() == entity_type ): return schema['id'] for schema in self.session.schemas: if schema['id'].lower() == entity_type: return schema['id'] raise ValueError( 'Unable to translate entity type: {0}.'.format(entity_type) ) def _launch(self, event): args = self._translate_event( self.session, event ) interface = self._interface( self.session, *args ) if interface: return interface response = self.launch( self.session, *args ) return self._handle_result( self.session, response, *args )
[docs] def launch(self, session, entities, event): '''Callback method for the custom action. return either a bool ( True if successful or False if the action failed ) or a dictionary with they keys `message` and `success`, the message should be a string and will be displayed as feedback to the user, success should be a bool, True if successful or False if the action failed. *session* is a `ftrack_api.Session` instance *entities* is a list of tuples each containing the entity type and the entity id. If the entity is a hierarchical you will always get the entity type TypedContext, once retrieved through a get operation you will have the "real" entity type ie. example Shot, Sequence or Asset Build. *event* the unmodified original event ''' raise NotImplementedError()
def _interface(self, *args): interface = self.interface(*args) if interface: return { 'items': interface }
[docs] def interface(self, session, entities, event): '''Return a interface if applicable or None *session* is a `ftrack_api.Session` instance *entities* is a list of tuples each containing the entity type and the entity id. If the entity is a hierarchical you will always get the entity type TypedContext, once retrieved through a get operation you will have the "real" entity type ie. example Shot, Sequence or Asset Build. *event* the unmodified original event ''' return None
def _handle_result(self, session, result, entities, event): '''Validate the returned result from the action callback''' if isinstance(result, bool): result = { 'success': result, 'message': ( '{0} launched successfully.'.format( self.label ) ) } elif isinstance(result, dict): for key in ('success', 'message'): if key in result: continue raise KeyError( 'Missing required key: {0}.'.format(key) ) else: self.logger.error( 'Invalid result type must be bool or dictionary!' ) return result