Event listener: Restrict task status updates

This event listener can be used to limit which users are allowed to set certain statuses. When a status is changed in the ftrack interface, the event listener will check if the status is restricted, and if the user is member of the group with users that are allowed to set to those statuses. If not, the status will be reverted to the old value, and a message will inform the user that the operation was not permitted.

Configuration

To use the event listener, first set the two variables at the top:

RESTRICTED_STATUSES: A list of task status names that are restricted, only the members of a group will be allowed to change these.
ALLOWED_GROUP_NAME: The name of the group in ftrack which contains the users that are allowed to set the restricted statuses.

Warning

The API user running the event listner must be a member of the group, otherwise it might end up in an infinite loop when changing between two restricted statuses.

Running the event listener

You can run the event listener from the command line:

python restrict_task_status_listener.py

You can also register the event listener as a plugin. For more information, see the links below.

Tip

You can use something like Supervisor or Monit to monitor and ensure the event listener is always running.

Find more information in our documentaion:

restrict_task_status_listener.py

Raw

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

# This snippet was last tested with ftrack 3.3.31 and Python API 0.15.5.

import logging
import ftrack_api
import functools

logger = logging.getLogger('restrict_task_status_listener')

RESTRICTED_STATUSES = [
    'Approved'
]
ALLOWED_GROUP_NAME = 'Producers'

def restrict_status_updates(session, group_id, restricted_status_ids, event):
    '''Handle *event*.'''
    status_reverted = False
    user_id = None

    for entity in event['data'].get('entities', []):
        if (
            entity.get('entityType') == 'task' and
            entity.get('action') in ('update',) and
            'statusid' in entity.get('keys', [])
        ):
            entity_id = entity['entityId']
            user_id = event['source'].get('user', {}).get('id', None)
            if not user_id:
                logger.warning('No source user, allowing update...')
                continue

            old_status_id = entity['changes']['statusid']['old']
            new_status_id = entity['changes']['statusid']['new']
            if not new_status_id in restricted_status_ids:
                logger.info('Status is not restricted, allowing update...')
                continue

            # Make sure user is a member in the allowed group.
            membership = session.query(
                'Membership where user_id is "{}" and group_id is "{}"'.format(
                    user_id, group_id
                )
            ).first()

            if not membership:
                logger.info('Status change is _not_ allowed, reverting')
                status_reverted = True
                task = session.get('Task', entity_id)
                task['status_id'] = old_status_id
            else:
                logger.info('Status change is allowed')

    if status_reverted:
        # Persist changes
        try:
            session.commit()
        except Exception:
            session.rollback()
            raise

        # Trigger a message to the user (new in ftrack 3.3.31)
        session.event_hub.publish(
            ftrack_api.event.base.Event(
                topic='ftrack.action.trigger-user-interface',
                data=dict(
                    type='message',
                    success=False,
                    message='You are not allowed to set this status'
                ),
                target='applicationId=ftrack.client.web and user.id="{0}"'.format(user_id)
            ),
            on_error='ignore'
        )


def register(session, **kw):
    '''Register event listener.'''

    # Validate that session is an instance of ftrack_api.Session. If not,
    # assume that register is being called from an incompatible API
    # and return without doing anything.
    if not isinstance(session, ftrack_api.Session):
        # Exit to avoid registering this plugin again.
        return

    # Get the ID of the allowed group 
    allowed_group = session.query(
        'Group where name is {0}'.format(ALLOWED_GROUP_NAME)
    ).one()
    allowed_group_id = allowed_group['id']

    # Get the ids of the restricted statuses.
    restricted_statuses = session.query(
        'Status where name in ("{0}")'.format(
            '", "'.join(RESTRICTED_STATUSES)
        )
    )
    restricted_status_ids = [status['id'] for status in restricted_statuses]

    # Register the event handler
    handle_event = functools.partial(
         restrict_status_updates, session, allowed_group_id,
         restricted_status_ids
    )
    session.event_hub.subscribe('topic=ftrack.update', handle_event)


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    session = ftrack_api.Session()
    register(session)

    # Wait for events.
    session.event_hub.wait()

    

Comments (5)

HTTPS

SSH

You can clone a snippet to your computer for local editing. Learn more.

Snippets

ftrack Event listener - Restrict task status updates

Event listener: Restrict task status updates

Configuration

Running the event listener

Read more

Comments (5)