Ftrack server action improvement

pype/modules/ftrack/__init__.py

@@ -1,12 +1,13 @@
 from . import ftrack_server
 from .ftrack_server import FtrackServer, check_ftrack_url
-from .lib import BaseHandler, BaseEvent, BaseAction
+from .lib import BaseHandler, BaseEvent, BaseAction, ServerAction
 
 __all__ = (
  "ftrack_server",
  "FtrackServer",
  "check_ftrack_url",
  "BaseHandler",
  "BaseEvent",
- "BaseAction"
+ "BaseAction",
+ "ServerAction"
 )

pype/modules/ftrack/events/action_push_frame_values_to_task.py

@@ -1,10 +1,10 @@
 import json
 import collections
 import ftrack_api
-from pype.modules.ftrack.lib import BaseAction
+from pype.modules.ftrack.lib import ServerAction
 
 
-class PushFrameValuesToTaskAction(BaseAction):
+class PushFrameValuesToTaskAction(ServerAction):
  """Action for testing purpose or as base for new actions."""
 
  # Ignore event handler by default
@@ -34,50 +34,14 @@ class PushFrameValuesToTaskAction(BaseAction):
  "frameStart": "fstart",
  "frameEnd": "fend"
  }
- discover_role_list = {"Pypeclub", "Administrator", "Project Manager"}
-
- def register(self):
- modified_role_names = set()
- for role_name in self.discover_role_list:
- modified_role_names.add(role_name.lower())
- self.discover_role_list = modified_role_names
-
- self.session.event_hub.subscribe(
- "topic=ftrack.action.discover",
- self._discover,
- priority=self.priority
- )
-
- launch_subscription = (
- "topic=ftrack.action.launch and data.actionIdentifier={0}"
- ).format(self.identifier)
- self.session.event_hub.subscribe(launch_subscription, self._launch)
+ role_list = {"Pypeclub", "Administrator", "Project Manager"}
 
  def discover(self, session, entities, event):
  """ Validation """
  # Check if selection is valid
- valid_selection = False
  for ent in event["data"]["selection"]:
  # Ignore entities that are not tasks or projects
  if ent["entityType"].lower() == "show":
- valid_selection = True
- break
-
- if not valid_selection:
- return False
-
- # Get user and check his roles
- user_id = event.get("source", {}).get("user", {}).get("id")
- if not user_id:
- return False
-
- user = session.query("User where id is \"{}\"".format(user_id)).first()
- if not user:
- return False
-
- for role in user["user_security_roles"]:
- lowered_role = role["security_role"]["name"].lower()
- if lowered_role in self.discover_role_list:
  return True
  return False
 

pype/modules/ftrack/events/action_sync_to_avalon.py

@@ -1,11 +1,11 @@
 import time
 import traceback
 
-from pype.modules.ftrack import BaseAction
+from pype.modules.ftrack import ServerAction
 from pype.modules.ftrack.lib.avalon_sync import SyncEntitiesFactory
 
 
-class SyncToAvalonServer(BaseAction):
+class SyncToAvalonServer(ServerAction):
  """
  Synchronizing data action - from Ftrack to Avalon DB
 
@@ -36,48 +36,18 @@ class SyncToAvalonServer(BaseAction):
  variant = "- Sync To Avalon (Server)"
  #: Action description.
  description = "Send data from Ftrack to Avalon"
+ role_list = {"Pypeclub", "Administrator", "Project Manager"}
 
  def __init__(self, *args, **kwargs):
  super().__init__(*args, **kwargs)
  self.entities_factory = SyncEntitiesFactory(self.log, self.session)
 
- def register(self):
- self.session.event_hub.subscribe(
- "topic=ftrack.action.discover",
- self._discover,
- priority=self.priority
- )
-
- launch_subscription = (
- "topic=ftrack.action.launch and data.actionIdentifier={0}"
- ).format(self.identifier)
- self.session.event_hub.subscribe(launch_subscription, self._launch)
-
  def discover(self, session, entities, event):
  """ Validation """
  # Check if selection is valid
- valid_selection = False
  for ent in event["data"]["selection"]:
  # Ignore entities that are not tasks or projects
  if ent["entityType"].lower() in ["show", "task"]:
- valid_selection = True
- break
-
- if not valid_selection:
- return False
-
- # Get user and check his roles
- user_id = event.get("source", {}).get("user", {}).get("id")
- if not user_id:
- return False
-
- user = session.query("User where id is \"{}\"".format(user_id)).first()
- if not user:
- return False
-
- role_list = ["Pypeclub", "Administrator", "Project Manager"]
- for role in user["user_security_roles"]:
- if role["security_role"]["name"] in role_list:
  return True
  return False
 

pype/modules/ftrack/lib/__init__.py

@@ -2,7 +2,7 @@
 from . import credentials
 from .ftrack_base_handler import BaseHandler
 from .ftrack_event_handler import BaseEvent
-from .ftrack_action_handler import BaseAction, statics_icon
+from .ftrack_action_handler import BaseAction, ServerAction, statics_icon
 from .ftrack_app_handler import AppAction
 
 __all__ = (
@@ -11,6 +11,7 @@
  "BaseHandler",
  "BaseEvent",
  "BaseAction",
+ "ServerAction",
  "statics_icon",
  "AppAction"
 )

pype/modules/ftrack/lib/ftrack_action_handler.py

@@ -195,3 +195,82 @@ def _handle_result(self, result):
  ).format(str(type(result))))
 
  return result
+
+
+class ServerAction(BaseAction):
+ """Action class meant to be used on event server.
+
+ Unlike the `BaseAction` roles are not checked on register but on discover.
+ For the same reason register is modified to not filter topics by username.
+ """
+
+ def __init__(self, *args, **kwargs):
+ if not self.role_list:
+ self.role_list = set()
+ else:
+ self.role_list = set(
+ role_name.lower()
+ for role_name in self.role_list
+ )
+ super(ServerAction, self).__init__(*args, **kwargs)
+
+ def _register_role_check(self):
+ # Skip register role check.
+ return
+
+ def _discover(self, event):
+ """Check user discover availability."""
+ if not self._check_user_discover(event):
+ return
+ return super(ServerAction, self)._discover(event)
+
+ def _check_user_discover(self, event):
+ """Should be action discovered by user trying to show actions."""
+ if not self.role_list:
+ return True
+
+ user_entity = self._get_user_entity(event)
+ if not user_entity:
+ return False
+
+ for role in user_entity["user_security_roles"]:
+ lowered_role = role["security_role"]["name"].lower()
+ if lowered_role in self.role_list:
+ return True
+ return False
+
+ def _get_user_entity(self, event):
+ """Query user entity from event."""
+ not_set = object()
+
+ # Check if user is already stored in event data
+ user_entity = event["data"].get("user_entity", not_set)
+ if user_entity is not_set:
+ # Query user entity from event
+ user_info = event.get("source", {}).get("user", {})
+ user_id = user_info.get("id")
+ username = user_info.get("username")
+ if user_id:
+ user_entity = self.session.query(
+ "User where id is {}".format(user_id)
+ ).first()
+ if not user_entity and username:
+ user_entity = self.session.query(
+ "User where username is {}".format(username)
+ ).first()
+ event["data"]["user_entity"] = user_entity
+
+ return user_entity
+
+ def register(self):
+ """Register subcription to Ftrack event hub."""
+ self.session.event_hub.subscribe(
+ "topic=ftrack.action.discover",
+ self._discover,
+ priority=self.priority
+ )
+
+ launch_subscription = (
+ "topic=ftrack.action.launch and data.actionIdentifier={0}"
+ ).format(self.identifier)
+ self.session.event_hub.subscribe(launch_subscription, self._launch)

pype/modules/ftrack/lib/ftrack_base_handler.py

@@ -35,6 +35,7 @@ class BaseHandler(object):
  type = 'No-type'
  ignore_me = False
  preactions = []
+ role_list = []
 
  def __init__(self, session, plugins_presets=None):
  '''Expects a ftrack_api.Session instance'''
@@ -148,20 +149,27 @@ def session(self):
  def reset_session(self):
  self.session.reset()
 
+ def _register_role_check(self):
+ if not self.role_list or not isinstance(self.role_list, (list, tuple)):
+ return
+
+ user_entity = self.session.query(
+ "User where username is \"{}\"".format(self.session.api_user)
+ ).one()
+ available = False
+ lowercase_rolelist = [
+ role_name.lower()
+ for role_name in self.role_list
+ ]
+ for role in user_entity["user_security_roles"]:
+ if role["security_role"]["name"].lower() in lowercase_rolelist:
+ available = True
+ break
+ if available is False:
+ raise MissingPermision
+
  def _preregister(self):
- if hasattr(self, "role_list") and len(self.role_list) > 0:
- username = self.session.api_user
- user = self.session.query(
- 'User where username is "{}"'.format(username)
- ).one()
- available = False
- lowercase_rolelist = [x.lower() for x in self.role_list]
- for role in user['user_security_roles']:
- if role['security_role']['name'].lower() in lowercase_rolelist:
- available = True
- break
- if available is False:
- raise MissingPermision
+ self._register_role_check()
 
  # Custom validations
  result = self.preregister()
@@ -172,12 +180,11 @@ def _preregister(self):
  ).format(self.__class__.__name__))
  return
 
- if result is True:
- return
- msg = None
- if isinstance(result, str):
- msg = result
- raise PreregisterException(msg)
+ if result is not True:
+ msg = None
+ if isinstance(result, str):
+ msg = result
+ raise PreregisterException(msg)
 
  def preregister(self):
  '''