Bases: FtrackPublishContextPlugin
Collect custom attribute values and store them to customData.
Data are stored into each instance in context under instance.data["customData"]["ftrack"].
Hierarchical attributes are not looked up properly for that functionality custom attribute values lookup must be extended.
Source code in client/ayon_ftrack/plugins/publish/collect_custom_attributes_data.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
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
148
149 | class CollectFtrackCustomAttributeData(plugin.FtrackPublishContextPlugin):
"""Collect custom attribute values and store them to customData.
Data are stored into each instance in context under
instance.data["customData"]["ftrack"].
Hierarchical attributes are not looked up properly for that functionality
custom attribute values lookup must be extended.
"""
order = pyblish.api.CollectorOrder + 0.4992
label = "Collect ftrack Custom Attribute Data"
# Name of custom attributes for which will be look for
custom_attribute_keys = []
def process(self, context):
if not self.custom_attribute_keys:
self.log.info("Custom attribute keys are not set. Skipping")
return
ftrack_entities_by_id = {}
default_entity_id = None
context_entity = context.data.get("ftrackEntity")
if context_entity:
entity_id = context_entity["id"]
default_entity_id = entity_id
ftrack_entities_by_id[entity_id] = context_entity
instances_by_entity_id = {
default_entity_id: []
}
for instance in context:
entity = instance.data.get("ftrackEntity")
if not entity:
instances_by_entity_id[default_entity_id].append(instance)
continue
entity_id = entity["id"]
ftrack_entities_by_id[entity_id] = entity
if entity_id not in instances_by_entity_id:
instances_by_entity_id[entity_id] = []
instances_by_entity_id[entity_id].append(instance)
if not ftrack_entities_by_id:
self.log.info("ftrack entities are not set. Skipping")
return
session = context.data["ftrackSession"]
custom_attr_key_by_id = self.query_attr_confs(session)
if not custom_attr_key_by_id:
self.log.info((
"Didn't find any of defined custom attributes {}"
).format(", ".join(self.custom_attribute_keys)))
return
entity_ids = list(instances_by_entity_id.keys())
values_by_entity_id = self.query_attr_values(
session, entity_ids, custom_attr_key_by_id
)
for entity_id, instances in instances_by_entity_id.items():
if entity_id not in values_by_entity_id:
# Use default empty values
entity_id = None
for instance in instances:
value = copy.deepcopy(values_by_entity_id[entity_id])
if "customData" not in instance.data:
instance.data["customData"] = {}
instance.data["customData"]["ftrack"] = value
instance_label = (
instance.data.get("label") or instance.data["name"]
)
self.log.debug((
"Added ftrack custom data to instance \"{}\": {}"
).format(instance_label, value))
def query_attr_values(self, session, entity_ids, custom_attr_key_by_id):
# Prepare values for query
entity_ids_joined = ",".join([
'"{}"'.format(entity_id)
for entity_id in entity_ids
])
conf_ids_joined = ",".join([
'"{}"'.format(conf_id)
for conf_id in custom_attr_key_by_id.keys()
])
# Query custom attribute values
value_items = session.query(
(
"select value, entity_id, configuration_id"
" from CustomAttributeValue"
" where entity_id in ({}) and configuration_id in ({})"
).format(
entity_ids_joined,
conf_ids_joined
)
).all()
# Prepare default value output per entity id
values_by_key = {
key: None for key in self.custom_attribute_keys
}
# Prepare all entity ids that were queried
values_by_entity_id = {
entity_id: copy.deepcopy(values_by_key)
for entity_id in entity_ids
}
# Add none entity id which is used as default value
values_by_entity_id[None] = copy.deepcopy(values_by_key)
# Go through queried data and store them
for item in value_items:
conf_id = item["configuration_id"]
conf_key = custom_attr_key_by_id[conf_id]
entity_id = item["entity_id"]
values_by_entity_id[entity_id][conf_key] = item["value"]
return values_by_entity_id
def query_attr_confs(self, session):
custom_attributes = set(self.custom_attribute_keys)
cust_attrs_query = (
"select id, key from CustomAttributeConfiguration"
" where key in ({})"
).format(", ".join(
["\"{}\"".format(attr_name) for attr_name in custom_attributes]
))
custom_attr_confs = session.query(cust_attrs_query).all()
return {
conf["id"]: conf["key"]
for conf in custom_attr_confs
}
|