Skip to content

custom_attributes

default_custom_attributes_definition()

Default custom attribute definitions created in ftracl.

Todos

Convert to list of dictionaries to be able determine order. Check if ftrack api support to define order first!

Returns:

Type Description

dict[str, Any]: Custom attribute configurations per entity type that can be used to create/update custom attributes.

Source code in client/ayon_ftrack/common/custom_attributes.py
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
def default_custom_attributes_definition():
    """Default custom attribute definitions created in ftracl.

    Todos:
        Convert to list of dictionaries to be able determine order. Check if
            ftrack api support to define order first!

    Returns:
        dict[str, Any]: Custom attribute configurations per entity type that
            can be used to create/update custom attributes.

    """
    # TODO use AYON built-in attributes as source of truth
    json_file_path = os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        "custom_attributes.json"
    )
    with open(json_file_path, "r") as json_stream:
        data = json.load(json_stream)
    return data

ensure_custom_attribute_group_exists(session, group, groups=None)

Ensure custom attribute group in ftrack.

Parameters:

Name Type Description Default
session Session

Connected ftrack session.

required
group str

Name of group.

required
groups Optional[List[Entity]]

Pre-fetched custom attribute groups.

None

Returns:

Name Type Description
FtrackEntity Entity

Created custom attribute group.

Source code in client/ayon_ftrack/common/custom_attributes.py
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def ensure_custom_attribute_group_exists(
    session: ftrack_api.Session,
    group: str,
    groups: Optional[List["FtrackEntity"]] = None,
) -> "FtrackEntity":
    """Ensure custom attribute group in ftrack.

    Args:
        session (ftrack_api.Session): Connected ftrack session.
        group (str): Name of group.
        groups (Optional[List[FtrackEntity]]): Pre-fetched
            custom attribute groups.

    Returns:
        FtrackEntity: Created custom attribute group.

    """
    if groups is None:
        groups = session.query(
            "select id, name from CustomAttributeGroup"
        ).all()
    low_name = group.lower()
    for group in groups:
        if group["name"].lower() == low_name:
            return group

    group = session.create(
        "CustomAttributeGroup",
        {"name": group}
    )
    session.commit()
    return group

ensure_mandatory_custom_attributes_exists(session, addon_settings, attr_confs=None, custom_attribute_types=None, groups=None, security_roles=None)

Make sure that mandatory custom attributes exists in ftrack.

Parameters:

Name Type Description Default
session Session

Connected ftrack session.

required
addon_settings Dict[str, Any]

Addon settings.

required
attr_confs Optional[List[Entity]]

Pre-fetched all existing custom attribute configurations in ftrack.

None
custom_attribute_types Optional[List[Entity]]

Pre-fetched custom attribute types.

None
groups Optional[List[Entity]]

Pre-fetched custom attribute groups.

None
security_roles Optional[List[Entity]]

Pre-fetched security roles.

None
Source code in client/ayon_ftrack/common/custom_attributes.py
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
def ensure_mandatory_custom_attributes_exists(
    session: ftrack_api.Session,
    addon_settings: Dict[str, Any],
    attr_confs: Optional[List["FtrackEntity"]] = None,
    custom_attribute_types: Optional[List["FtrackEntity"]] = None,
    groups: Optional[List["FtrackEntity"]] = None,
    security_roles: Optional[List["FtrackEntity"]] = None,
):
    """Make sure that mandatory custom attributes exists in ftrack.

    Args:
        session (ftrack_api.Session): Connected ftrack session.
        addon_settings (Dict[str, Any]): Addon settings.
        attr_confs (Optional[List[FtrackEntity]]): Pre-fetched all existing
            custom attribute configurations in ftrack.
        custom_attribute_types (Optional[List[FtrackEntity]]): Pre-fetched
            custom attribute types.
        groups (Optional[List[FtrackEntity]]): Pre-fetched custom attribute
            groups.
        security_roles (Optional[List[FtrackEntity]]): Pre-fetched security
            roles.

    """
    if attr_confs is None:
        attr_confs = get_all_attr_configs(session)

    # Split existing custom attributes
    attr_confs_by_entity_type = collections.defaultdict(list)
    hier_confs = []
    for attr_conf in attr_confs:
        if attr_conf["is_hierarchical"]:
            hier_confs.append(attr_conf)
        else:
            entity_type = attr_conf["entity_type"]
            attr_confs_by_entity_type[entity_type].append(attr_conf)

    # Prepare possible attribute types
    if custom_attribute_types is None:
        custom_attribute_types = session.query(
            "select id, name from CustomAttributeType"
        ).all()

    attr_type_id_by_low_name = {
        attr_type["name"].lower(): attr_type["id"]
        for attr_type in custom_attribute_types
    }

    if security_roles is None:
        security_roles = session.query(
            "select id, name, type from SecurityRole"
        ).all()

    security_roles = {
        role["name"].lower(): role
        for role in security_roles
    }
    mandatory_attributes_settings = (
        addon_settings
        ["custom_attributes"]
        ["mandatory_attributes"]
    )

    # Prepare group
    group_entity = ensure_custom_attribute_group_exists(
        session, CUST_ATTR_GROUP, groups
    )
    group_id = group_entity["id"]

    for item in [
        {
            "key": CUST_ATTR_KEY_SERVER_ID,
            "type": "text",
            "label": "AYON ID",
            "default": "",
            "is_hierarchical": True,
            "config": {"markdown": False},
            "group_id": group_id,
        },
        {
            "key": CUST_ATTR_KEY_SERVER_PATH,
            "type": "text",
            "label": "AYON path",
            "default": "",
            "is_hierarchical": True,
            "config": {"markdown": False},
            "group_id": group_id,
        },
        {
            "key": CUST_ATTR_KEY_SYNC_FAIL,
            "type": "boolean",
            "label": "AYON sync failed",
            "is_hierarchical": True,
            "default": False,
            "group_id": group_id,
        },
        {
            "key": CUST_ATTR_AUTO_SYNC,
            "type": "boolean",
            "label": "AYON auto-sync",
            "default": False,
            "is_hierarchical": False,
            "entity_type": "show",
            "group_id": group_id,
        }
    ]:
        key = item["key"]
        attr_settings = mandatory_attributes_settings[key]
        read_roles = []
        write_roles = []
        for role_names, roles in (
            (attr_settings["read_security_roles"], read_roles),
            (attr_settings["write_security_roles"], write_roles),
        ):
            if not role_names:
                roles.extend(security_roles.values())
                continue

            for name in role_names:
                role = security_roles.get(name.lower())
                if role is not None:
                    roles.append(role)

        is_hierarchical = item["is_hierarchical"]
        entity_type_confs = hier_confs
        if not is_hierarchical:
            entity_type = item["entity_type"]
            entity_type_confs = attr_confs_by_entity_type.get(entity_type, [])
        matching_attr_conf = next(
            (
                attr_conf
                for attr_conf in entity_type_confs
                if attr_conf["key"] == key
            ),
            None
        )

        entity_data = copy.deepcopy(item)
        attr_type = entity_data.pop("type")
        entity_data["type_id"] = attr_type_id_by_low_name[attr_type.lower()]
        # Convert 'config' to json string
        config = entity_data.get("config")
        if isinstance(config, dict):
            entity_data["config"] = json.dumps(config)

        if matching_attr_conf is None:
            # Make sure 'entity_type' is filled for hierarchical attribute
            # - it is required to be able to create custom attribute
            if is_hierarchical:
                entity_data.setdefault("entity_type", "show")
            # Make sure config is set to empty dictionary for creation
            entity_data.setdefault("config", "{}")
            entity_data["read_security_roles"] = read_roles
            entity_data["write_security_roles"] = write_roles
            session.create(
                "CustomAttributeConfiguration",
                entity_data
            )
            session.commit()
            continue

        changed = False
        for key, value in entity_data.items():
            if matching_attr_conf[key] != value:
                matching_attr_conf[key] = value
                changed = True

        match_read_role_ids = {
            role["id"] for role in matching_attr_conf["read_security_roles"]
        }
        match_write_role_ids = {
            role["id"] for role in matching_attr_conf["write_security_roles"]
        }
        if match_read_role_ids != {role["id"] for role in read_roles}:
            matching_attr_conf["read_security_roles"] = read_roles
            changed = True
        if match_write_role_ids != {role["id"] for role in write_roles}:
            matching_attr_conf["write_security_roles"] = write_roles
            changed = True

        if changed:
            session.commit()

get_all_attr_configs(session, fields=None)

Query custom attribute configurations from ftrack server.

Parameters:

Name Type Description Default
session Session

Connected ftrack session.

required
fields Optional[Iterable[str]]

Field to query for attribute configurations.

None

Returns:

Type Description
List[Entity]

List[FtrackEntity]: ftrack custom attributes.

Source code in client/ayon_ftrack/common/custom_attributes.py
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
def get_all_attr_configs(
    session: ftrack_api.Session,
    fields: Optional[Iterable[str]] = None,
) -> List["FtrackEntity"]:
    """Query custom attribute configurations from ftrack server.

    Args:
        session (ftrack_api.Session): Connected ftrack session.
        fields (Optional[Iterable[str]]): Field to query for
            attribute configurations.

    Returns:
        List[FtrackEntity]: ftrack custom attributes.

    """
    if not fields:
        fields = {
            "id",
            "key",
            "entity_type",
            "object_type_id",
            "is_hierarchical",
            "default",
            "group_id",
            "type_id",
            # "config",
            # "label",
            # "sort",
            # "project_id",
        }

    joined_fields = ", ".join(set(fields))

    return session.query(
        f"select {joined_fields} from CustomAttributeConfiguration"
    ).all()

get_custom_attributes_by_entity_id(session, entity_ids, attr_configs, skip_none_values=True, store_by_key=True)

Query custom attribute values and store their value by entity and attr.

There is option to return values by attribute key or attribute id. In case the output should be stored by key and there is hierarchical attribute with same key as non-hierarchical it's then hierarchical value has priority of usage.

Parameters:

Name Type Description Default
session Session

Connected ftrack session.

required
entity_ids Iterable[str]

Entity ids for which custom attribute values should be returned.

required
attr_configs

Custom attribute configurations.

required
skip_none_values bool

Custom attribute with value set to 'None' won't be in output.

True
store_by_key bool

Output will be stored by attribute keys if true otherwise is value stored by attribute id.

True

Returns:

Type Description

Dict[str, Dict[str, Any]]: Custom attribute values by entity id.

Source code in client/ayon_ftrack/common/custom_attributes.py
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
def get_custom_attributes_by_entity_id(
    session,
    entity_ids,
    attr_configs,
    skip_none_values=True,
    store_by_key=True
):
    """Query custom attribute values and store their value by entity and attr.

    There is option to return values by attribute key or attribute id. In case
    the output should be stored by key and there is hierarchical attribute
    with same key as non-hierarchical it's then hierarchical value
    has priority of usage.

    Args:
        session (ftrack_api.Session): Connected ftrack session.
        entity_ids (Iterable[str]): Entity ids for which custom attribute
            values should be returned.
        attr_configs: Custom attribute configurations.
        skip_none_values (bool): Custom attribute with value set to 'None'
            won't be in output.
        store_by_key (bool): Output will be stored by attribute keys if true
            otherwise is value stored by attribute id.

    Returns:
        Dict[str, Dict[str, Any]]: Custom attribute values by entity id.

    """
    entity_ids = set(entity_ids)
    hier_attr_ids = {
        attr_conf["id"]
        for attr_conf in attr_configs
        if attr_conf["is_hierarchical"]
    }
    attr_by_id = {
        attr_conf["id"]: attr_conf["key"]
        for attr_conf in attr_configs
    }

    value_items = query_custom_attribute_values(
        session, attr_by_id.keys(), entity_ids
    )

    output = collections.defaultdict(dict)
    for value_item in value_items:
        value = value_item["value"]
        if skip_none_values and value is None:
            continue

        entity_id = value_item["entity_id"]
        entity_values = output[entity_id]
        attr_id = value_item["configuration_id"]
        if not store_by_key:
            entity_values[attr_id] = value
            continue

        attr_key = attr_by_id[attr_id]
        # Hierarchical attributes are always preferred
        if attr_key not in entity_values or attr_id in hier_attr_ids:
            entity_values[attr_key] = value

    return output

get_custom_attributes_mapping(session, addon_settings, attr_confs=None, ayon_attributes=None)

Query custom attribute configurations from ftrack server.

Returns:

Type Description
CustomAttributesMapping

Dict[str, List[object]]: ftrack custom attributes.

Source code in client/ayon_ftrack/common/custom_attributes.py
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def get_custom_attributes_mapping(
    session: ftrack_api.Session,
    addon_settings: Dict[str, Any],
    attr_confs: Optional[List[object]] = None,
    ayon_attributes: Optional[List[object]] = None,
) -> CustomAttributesMapping:
    """Query custom attribute configurations from ftrack server.

    Returns:
        Dict[str, List[object]]: ftrack custom attributes.

    """
    cust_attr = addon_settings["custom_attributes"]
    # "custom_attributes/attributes_mapping/mapping"
    attributes_mapping = cust_attr["attributes_mapping"]

    if attr_confs is None:
        attr_confs = get_all_attr_configs(session)

    if ayon_attributes is None:
        ayon_attributes = ayon_api.get_attributes_schema()["attributes"]

    ayon_attribute_names = {
        attr["name"]
        for attr in ayon_attributes
    }

    hier_attrs = []
    nonhier_attrs = []
    for attr_conf in attr_confs:
        if attr_conf["is_hierarchical"]:
            hier_attrs.append(attr_conf)
        else:
            nonhier_attrs.append(attr_conf)

    output = CustomAttributesMapping()
    if not attributes_mapping["enabled"]:
        builtin_attrs = {
            attr["name"]
            for attr in ayon_attributes
            if attr["builtin"]
        }
        for attr_conf in hier_attrs:
            attr_name = attr_conf["key"]
            # Use only AYON attribute hierarchical equivalent
            if (
                attr_name in output
                or attr_name not in ayon_attribute_names
            ):
                continue

            # Attribute must be in builtin attributes or openpype/ayon group
            # NOTE get rid of group name check when only mapping is used
            if (
                attr_name in builtin_attrs
                or attr_conf["group"]["name"] in ("openpype", CUST_ATTR_GROUP)
            ):
                output.add_mapping_item(MappedAYONAttribute(
                    attr_name,
                    True,
                    [attr_conf],
                ))

    else:
        for item in attributes_mapping["mapping"]:
            ayon_attr_name = item["name"]
            if ayon_attr_name not in ayon_attribute_names:
                continue

            is_hierarchical = item["attr_type"] == "hierarchical"

            mapped_item = MappedAYONAttribute(
                ayon_attr_name, is_hierarchical, []
            )

            if is_hierarchical:
                attr_name = item["hierarchical"]
                for attr_conf in hier_attrs:
                    if attr_conf["key"] == attr_name:
                        mapped_item.add_attr_conf(attr_conf)
                        break
            else:
                attr_names = item["standard"]
                for attr_conf in nonhier_attrs:
                    if attr_conf["key"] in attr_names:
                        mapped_item.add_attr_conf(attr_conf)
            output.add_mapping_item(mapped_item)

    for attr_name in ayon_attribute_names:
        if attr_name not in output:
            output.add_mapping_item(MappedAYONAttribute(attr_name))

    return output

query_custom_attribute_values(session, attr_ids, entity_ids)

Query custom attribute values from ftrack database.

Using ftrack call method result may differ based on used table name and version of ftrack server.

For hierarchical attributes you shou always use only_set_values=True otherwise result will be default value of custom attribute and it would not be possible to differentiate if value is set on entity or default value is used.

Parameters:

Name Type Description Default
session Session

Connected ftrack session.

required
attr_ids Iterable[str]

Attribute configuration ids.

required
entity_ids Iterable[str]

Entity ids for which are values queried.

required

Returns:

Type Description

List[Dict[str, Any]]: Results from server.

Source code in client/ayon_ftrack/common/custom_attributes.py
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def query_custom_attribute_values(session, attr_ids, entity_ids):
    """Query custom attribute values from ftrack database.

    Using ftrack call method result may differ based on used table name and
    version of ftrack server.

    For hierarchical attributes you shou always use `only_set_values=True`
    otherwise result will be default value of custom attribute and it would not
    be possible to differentiate if value is set on entity or default value is
    used.

    Args:
        session (ftrack_api.Session): Connected ftrack session.
        attr_ids (Iterable[str]): Attribute configuration ids.
        entity_ids (Iterable[str]): Entity ids for which are values queried.

    Returns:
        List[Dict[str, Any]]: Results from server.
    """

    output = []
    # Just skip
    attr_ids = set(attr_ids)
    entity_ids = set(entity_ids)
    if not attr_ids or not entity_ids:
        return output

    # Prepare values to query
    attributes_joined = join_filter_values(attr_ids)

    # Query values in chunks
    chunk_size = 5000 // len(attr_ids)
    # Make sure entity_ids is `list` for chunk selection
    for chunk in create_chunks(entity_ids, chunk_size):
        entity_ids_joined = join_filter_values(chunk)
        output.extend(
            session.query(
                (
                    "select value, entity_id, configuration_id"
                    " from CustomAttributeValue"
                    " where entity_id in ({}) and configuration_id in ({})"
                ).format(entity_ids_joined, attributes_joined)
            ).all()
        )
    return output