Frame13/ayon-core
1
0
Fork 1
mirror of https://github.com/ynput/ayon-core.git synced 2025-12-26 05:42:15 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #608 from pypeclub/feature/ftrack_server_action

Browse source 
Ftrack server action improvement
This commit is contained in:
Milan Kolar 2020-10-08 12:10:37 +02:00 • committed by GitHub
commit e032feaf1e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 116 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

5
pype/modules/ftrack/__init__.py
View file

@ -1,6 +1,6 @@
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",
@ -8,5 +8,6 @@ __all__ = (
"check_ftrack_url",
"BaseHandler",
"BaseEvent",
"BaseAction"
"BaseAction",
"ServerAction"
)

42
pype/modules/ftrack/events/action_push_frame_values_to_task.py
View file

@ -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

36
pype/modules/ftrack/events/action_sync_to_avalon.py
View file

@ -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

3
pype/modules/ftrack/lib/__init__.py
View file

@ -2,7 +2,7 @@ from . import avalon_sync
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 @@ __all__ = (
"BaseHandler",
"BaseEvent",
"BaseAction",
"ServerAction",
"statics_icon",
"AppAction"
)

79
pype/modules/ftrack/lib/ftrack_action_handler.py
View file

@ -195,3 +195,82 @@ class BaseAction(BaseHandler):
).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)

45
pype/modules/ftrack/lib/ftrack_base_handler.py
View file

@ -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 @@ class BaseHandler(object):
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 @@ class BaseHandler(object):
).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):
'''