ftrack-query

ftrack-query

FTrack Query is an object-orientated wrapper over the FTrack API. While the default query syntax is powerful, it is entirely text based so dynamic queries can be difficult to construct. This module supports and/or operators with nested comparisons.

It is recommended to first read https://ftrack-python-api.readthedocs.io/en/stable/tutorial.html for a basic understanding of how the FTrack API works.

Installation

pip install ftrack_query

Information

Instead of writing the whole query string at once, a "statement" is constructed (eg. stmt = select('Task')), and the query can be built up by calling methods such as .where() and .populate() on the statement.

The CRUD methods are all supported (create, select, update, delete), but the main functionality is designed for use with select. The statements are built with a similar syntax to the main API so it should be straightforward to transition between the two.

Note that this is fully backwards compatible, so existing queries do not need to be rewritten.

Examples

The below example is for very basic queries:

from ftrack_query import FTrackQuery, attr, create, select, and_, or_

with FTrackQuery() as session:
    # Select
    project = session.select('Project').where(name='Test Project').one()

    # Create
    task = session.execute(
        create('Task').values(
            name='My Task',
            project=project,
        )
    )
    session.commit()

    # Update
    rows_updated = session.execute(
        update('Task')
        .where(name='Old Task Name')
        .values(name='New Task Name')
    )
    session.commit()

    # Delete
    rows_deleted = session.execute(
        delete('Task').where(
            name='Old Task Name',
        )
    )
    session.commit()

For a much more complex example:

ATTR_TYPE = attr('type.name')

TASK_STMT = (
    select('Task')
    # Filter the tasks
    .where(
        # Get any task without these statuses
        ~attr('status.name').in_(['Lighting', 'Rendering']),
        # Check for notes matching any of the following conditions:
        attr('notes').any(
            # Ensure note was posted by someone outside the company
            ~attr('user.email').endswith('@company.com')
            # Ensure note either not completable or not completed
            or_(
                and_(
                    completed_by=None,
                    is_todo=True,
                ),
                is_todo=False,
            ),
        ),
        # Ensure it has an animation task
        or_(
            ATTR_TYPE.contains('Animation'),
            ATTR_TYPE == 'Anim_Fixes',
        ),
    ),
    # Order the results
    .order_by(
        ATTR_TYPE.desc(),  # Equivalent to "type.name desc"
        'name',
    )
    # Use projections to populate attributes as part of the query
    .populate(
        'name',
        'notes',
        'status.name',
        ATTR_TYPE,
    )
    .limit(5)
)

with FTrackQuery() as session:
    # Filter the above query to the result of another query
    task_stmt = TASK_STMT.where(
        project_id=session.select('Project').where(name='Test Project').one()['id']
    )

    # Use the current session to execute the statement
    tasks = session.execute(task_stmt).all()

Events

The event system uses a slightly different query language.

from ftrack_query import FTrackQuery, event
from ftrack_query.event import attr, and_, or_

with FTrackQuery() as session:
    session.event_hub.subscribe(str(
        and_(
            attr('topic') == 'ftrack.update',
            attr('data.user.name') != getuser(),
        )
    ))
    session.event_hub.wait()

Note that attr(), and_(), and or_() are present in both ftrack_query and ftrack_query.event. These are not interchangable, so if both are needed, then import event and use that as the namespace.

API Reference

ftrack_query.FTrackQuery

Main session inherited from ftrack_api.Session.

ftrack_query.and_(*args, **kwargs) | ftrack_query.or_(*args, **kwargs)

Join multiple comparisons.

Shortcuts are provided with & and | (eg. attr(a).contains(b) & attr(x).contains(y)).

ftrack_query.not_(*args, **kwargs)

Reverse the input comparisons.

A shortcut is provided with ~ (eg. ~attr(x).contains(y)).

ftrack_query.select

Used for building the query string.

from ftrack_query import select

stmt = select(entity).where(...).populate(...)

Calling session.execute(stmt) will execute the query and return FTrack's own QueryResult object, from which .one(), .first() or .all() may be called. Alternatively, by using the shortcut session.select(entity), then this may be skipped.

where(*args, **kwargs)

Filter the result.

Using keywords is the fastest way, such as .where(first_name='Peter', last_name='Hunt'). For anything more complex than equality checks, using attr() is recommended, such as .where(attr('project.metadata').any(attr('key') != 'disabled')).

populate(*attrs)

Pre-fetch entity attributes.

An an example, in order to iterate through the name of every user, it would be a good idea to call .populate('first_name', 'last_name') as part of the query. Without that, it would take 2 separate queries per user, which is known as the N+1 query problem.

order_by(*attrs) | order(*attrs) | sort(*attrs)

Sort the results by an attribute.

The attribute and order can be given in the format attr('name').desc(), or as a raw string such as name descending. The order will default to ascending if not provided.

reverse()

Reverse the sorting direction.

limit(value)

Limit the amount of results to a certain value.

Note: This is incompatible with calling .first() or .one().

offset(value)

In the case of using a limit, apply an offset to the result that is returned.

options(**kwargs)

For advanced users only.

• page_size: Set the number of results to be fetched at once from FTrack.
• session: Attach a session object to the query.

subquery(attribute='id')

Make the statement a subquery for use within .in_().

This ensures there's always a "select from" as part of the statement. Manually setting the attribute parameter will override any existing projections.

ftrack_query.create

Used for creating new entities.

from ftrack_query import create

stmt = create(entity).values(...)

Calling session.execute(stmt) will return the created entity.

values(**kwargs)

Values to create the entity with.

ftrack_query.update

Used to batch update values on multiple entities. This is built off the select method so contains a lot of the same methods.

from ftrack_query import update

stmt = update(entity).where(...).values(...)

Calling session.execute(stmt) will return how many entities were found and updated.

where(*args, **kwargs)

Filter what to update.

values(**kwargs)

Values to update on the entity.

ftrack_query.delete

Used to delete entities. This is built off the select method so contains a lot of the same methods.

from ftrack_query import delete

stmt = delete(entity).where(...).options(remove_components=True)

Calling session.execute(stmt) will return how many entities were deleted.

where(*args, **kwargs)

Filter what to update.

options(**kwargs)

• remove_components: Remove any Component entity from every Location containing it before it is deleted. This is not enabled by default as it can't be rolled back.

ftrack_query.attr

The Comparison object is designed to convert data to a string. It contains a wide array of operators that can be used against any data type, including other Comparison objects. The function attr is a shortcut to this.

Any comparison can be reversed with the ~ prefix or the not_ function.

• String Comparison: attr(key) == 'value'
• Number comparison: attr(key) > 5
• Pattern Comparison: attr(key).like('value%')
• Time Comparison: attr(key).after(arrow.now().floor('day'))
• Scalar Relationship: attr(key).has(subkey='value')
• Collection Relationship: attr(key).any(subkey='value')
• Subquery Relationship: attr(key).in_(subquery)

__eq__(value) | __ne__(value) | __gt__(value) | __ge__(value) | __lt__(value) | __lt__(value)

Simple comparisons.

in_(values) | not_in(values)

Perform a check to check if an attribute matches any results.

This can accept a subquery such .in_('select id from table where x is y'), or a list of items like .in_(['x', 'y']).

like(value) | not_like(value) | startswith(value) | endwith(value) | contains(value)

Check if a string is contained within the query. Use a percent sign as the wildcard if using like or not_like; the rest are shortcuts and do this automatically.

has_(*args, **kwargs) | any_(*args, **kwargs)

Test against scalar and collection relationships.

before(values) | after(values)

Test against dates. Using arrow objects is recommended.

Equivalent examples from the API reference:

# Project
select('Project')

# Project where status is active
select('Project').where(status='active')

# Project where status is active and name like "%thrones"
select('Project').where(attr('name').like('%thrones'), status='active')

# session.query('Project where status is active and (name like "%thrones" or full_name like "%thrones")')
select('Project').where(or_(attr('name').like('%thrones'), attr('full_name').like('%thrones')), status='active')

# session.query('Task where project.id is "{0}"'.format(project['id']))
select('Task').where(project=project)

# session.query('Task where project.id is "{0}" and status.type.name is "Done"'.format(project['id']))
select('Task').where(attr('status.type.name') == 'Done', project=project)

# session.query('Task where timelogs.start >= "{0}"'.format(arrow.now().floor('day')))
select('Task').where(attr('timelogs.start') >= arrow.now().floor('day'))

# session.query('Note where author has (first_name is "Jane" and last_name is "Doe")')
select('Note').where(attr('author').has(first_name='Jane', last_name='Doe'))

# session.query('User where not timelogs any ()')
select('User').where(~attr('timelogs').any())

# projects = session.query('select full_name, status.name from Project')
select('Project').populate('full_name', 'status.name')

# select name from Project where allocations.resource[Group].memberships any (user.username is "john_doe")
select('Project').select('name').where(attr('allocations.resource[Group].memberships').any(attr('user.username') == 'john_doe'))

# Note where parent_id is "{version_id}" or parent_id in (select id from ReviewSessionObject where version_id is "{version_id}")
select('Note').where(or_(attr('parent_id').in_(select('ReviewSessionObject').where(version_id=version_id).subquery()), parent_id=version_id))