diff --git a/TODO.md b/TODO.md index e76efb6..8331158 100644 --- a/TODO.md +++ b/TODO.md @@ -1,9 +1,22 @@ +- Public API: + + Schema.is_variable_len(spec) -> can it return non-one results? + + Schema.resolve_entity(xxx) -> list of entity types + Schema.resolve_one_entity(xxx) -> one entity type or ValueError + + Schema.resolve_field(entity_type, field_spec) -> list of fields for entity_type + Schema.resolve_one_field(...) -> one field or ValueError + + - Create a standard-ish set of tags and aliases: $parent pointing to typical parent - Include inverse_association (from private schema) -- Schema.resolve_one -> error if there isn't just one - *type -> explicit prefix matching + +- basic key/value metadata store + - sgactions:ticket_project: 66 diff --git a/sgschema/schema.py b/sgschema/schema.py index 03f142b..3591f8b 100644 --- a/sgschema/schema.py +++ b/sgschema/schema.py @@ -1,4 +1,5 @@ import copy +import itertools import json import os import re @@ -176,39 +177,45 @@ def load(self, input_): if raw_schema: raise ValueError('unknown keys: %s' % ', '.join(sorted(raw_schema))) - def resolve(self, entity_spec, field_spec=None, auto_prefix=True, implicit_aliases=True, strict=False): + def resolve_entity(self, entity_spec, implicit_aliases=True, strict=False): - if field_spec is None: # We are resolving an entity. + op = entity_spec[0] + if op == '!': + return [entity_spec[1:]] + if op == '#': + return list(self.entity_tags.get(entity_spec[1:], ())) + if op == '$': + try: + return [self.entity_aliases[entity_spec[1:]]] + except KeyError: + return [] + if not op.isalnum(): + raise ValueError('unknown entity operation for %r' % entity_spec) - op = entity_spec[0] - if op == '!': - return [entity_spec[1:]] - if op == '#': - return list(self.entity_tags.get(entity_spec[1:], ())) - if op == '$': - try: - return [self.entity_aliases[entity_spec[1:]]] - except KeyError: - return [] - if not op.isalnum(): - raise ValueError('unknown entity operation for %r' % entity_spec) + if entity_spec in self.entities: + return [entity_spec] - if entity_spec in self.entities: - return [entity_spec] + if implicit_aliases and entity_spec in self.entity_aliases: + return [self.entity_aliases[entity_spec]] - if implicit_aliases and entity_spec in self.entity_aliases: - return [self.entity_aliases[entity_spec]] + if strict: + raise ValueError('%r is not an entity type' % entity_spec) - if strict: - raise ValueError('%r is not an entity' % entity_spec) + return [entity_spec] - return [entity_spec] + def resolve_one_entity(self, entity_spec, **kwargs): + res = self.resolve_entity(entity_spec, **kwargs) + if len(res) == 1: + return res[0] + else: + raise ValueError('%r returned %s entity types' % (entity_spec, len(res))) + + def _resolve_field(self, entity_spec, field_spec, auto_prefix=True, implicit_aliases=True, strict=False): - # When resolving a field, the entity must exist. try: entity = self.entities[entity_spec] except KeyError: - raise ValueError('%r is not an entity' % entity_spec) + raise ValueError('%r is not an entity type' % entity_spec) op = field_spec[0] if op == '!': @@ -239,6 +246,39 @@ def resolve(self, entity_spec, field_spec=None, auto_prefix=True, implicit_alias return [field_spec] + def resolve_field(self, entity_type, field_spec=None, auto_prefix=True, implicit_aliases=True, strict=False): + + spec_parts = field_spec.split('.') + + # Shortcut if there isn't anything fancy going on. + if len(spec_parts) == 1: + return self._resolve_field(entity_type, field_spec, auto_prefix, implicit_aliases, strict) + + # Crate list of [entity_type, field_spec] pairs. + spec_pairs = [[spec_parts[i-1] if i else None, spec_parts[i]] for i in xrange(0, len(spec_parts), 2)] + spec_pairs[0][0] = entity_type + + # Resolve each pair. + resolved_pair_sets = [] + for i, (entity_spec, field_spec), in enumerate(spec_pairs): + resolved_pairs = [] + resolved_pair_sets.append(resolved_pairs) + # Here entity types need not already exist; resolve them. + entity_types = self.resolve_entity(entity_spec, implicit_aliases, strict) + for entity_type in entity_types: + for field_name in self._resolve_field(entity_type, field_spec, auto_prefix, implicit_aliases, strict): + resolved_pairs.append((entity_type, field_name)) + + # Return the product of all resolved fields. + resolved_fields = [] + for pairs in itertools.product(*resolved_pair_sets): + field = '.'.join((entity_type + '.' if i else '') + field_name for i, (entity_type, field_name) in enumerate(pairs)) + resolved_fields.append(field) + return resolved_fields + + + + diff --git a/tests/__init__.py b/tests/__init__.py index d1005b7..883ba0f 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,3 +1,15 @@ +import os from unittest import TestCase from sgschema import Schema + + +def load_schema(name='mikeboers', raw=False): + path = os.path.abspath(os.path.join(__file__, '..', 'schema', name + ('.raw' if raw else '') + '.json')) + schema = Schema() + if raw: + schema.load_raw(path) + else: + schema.load(path) + return schema + diff --git a/tests/schema/dump.py b/tests/schema/dump.py new file mode 100644 index 0000000..51a0d45 --- /dev/null +++ b/tests/schema/dump.py @@ -0,0 +1,31 @@ + +import argparse +import urlparse +import os + +import shotgun_api3 + +import sgschema + + +parser = argparse.ArgumentParser() +parser.add_argument('-n', '--name') +parser.add_argument('base_url') +parser.add_argument('script_name') +parser.add_argument('api_key') +args = parser.parse_args() + + +shotgun = shotgun_api3.Shotgun(args.base_url, args.script_name, args.api_key) +schema = sgschema.Schema() +schema.read(shotgun) + + +if not args.name: + parsed = urlparse.urlparse(args.base_url) + args.name = parsed.netloc.split('.')[0] + +here = os.path.dirname(__file__) + +schema.dump(os.path.join(here, args.name + '.raw.json'), raw=True) +schema.dump(os.path.join(here, args.name + '.json')) diff --git a/tests/test_resolve.py b/tests/test_resolve.py index 737edc3..a3ccc32 100644 --- a/tests/test_resolve.py +++ b/tests/test_resolve.py @@ -22,26 +22,27 @@ def setUp(self): }) def test_explicit(self): - self.assertEqual(self.s.resolve('!Entity'), ['Entity']) - self.assertEqual(self.s.resolve('$A'), ['Entity']) - self.assertEqual(self.s.resolve('$B'), ['Entity']) - self.assertEqual(self.s.resolve('#X'), ['Entity']) - self.assertEqual(self.s.resolve('#Y'), ['Entity']) + self.assertEqual(self.s.resolve_entity('!Entity'), ['Entity']) + self.assertEqual(self.s.resolve_entity('$A'), ['Entity']) + self.assertEqual(self.s.resolve_entity('$B'), ['Entity']) + self.assertEqual(self.s.resolve_entity('#X'), ['Entity']) + self.assertEqual(self.s.resolve_entity('#Y'), ['Entity']) def test_namespace(self): - self.assertEqual(self.s.resolve('$with:Namespace'), ['Entity']) + self.assertEqual(self.s.resolve_entity('$with:Namespace'), ['Entity']) def test_implicit(self): - self.assertEqual(self.s.resolve('Entity'), ['Entity']) - self.assertEqual(self.s.resolve('A'), ['Entity']) - self.assertEqual(self.s.resolve('B'), ['Entity']) + self.assertEqual(self.s.resolve_entity('Entity'), ['Entity']) + self.assertEqual(self.s.resolve_entity('A'), ['Entity']) + self.assertEqual(self.s.resolve_entity('B'), ['Entity']) def test_missing(self): - self.assertEqual(self.s.resolve('#Missing'), []) - self.assertEqual(self.s.resolve('$Missing'), []) - self.assertEqual(self.s.resolve('!Missing'), ['Missing']) - self.assertEqual(self.s.resolve('Missing'), ['Missing']) - self.assertRaises(ValueError, self.s.resolve, 'Missing', strict=True) + self.assertEqual(self.s.resolve_entity('#Missing'), []) + self.assertEqual(self.s.resolve_entity('$Missing'), []) + self.assertEqual(self.s.resolve_entity('!Missing'), ['Missing']) + self.assertEqual(self.s.resolve_entity('Missing'), ['Missing']) + self.assertRaises(ValueError, self.s.resolve_entity, 'Missing', strict=True) + class TestResolveFields(TestCase): @@ -72,37 +73,90 @@ def setUp(self): }) def test_explicit(self): - self.assertEqual(self.s.resolve('Entity', '!attr'), ['attr']) - self.assertEqual(self.s.resolve('Entity', '$a'), ['attr']) - self.assertEqual(self.s.resolve('Entity', '$b'), ['attr']) - self.assertEqual(self.s.resolve('Entity', '#x'), ['attr']) - self.assertEqual(self.s.resolve('Entity', '#y'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '!attr'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '$a'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '$b'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '#x'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '#y'), ['attr']) def test_namespace(self): - self.assertEqual(self.s.resolve('Entity', '$with:namespace'), ['attr']) - self.assertEqual(self.s.resolve('Entity', 'with:namespace'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', '$with:namespace'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', 'with:namespace'), ['attr']) def test_implicit(self): - self.assertEqual(self.s.resolve('Entity', 'attr'), ['attr']) - self.assertEqual(self.s.resolve('Entity', 'a'), ['attr']) - self.assertEqual(self.s.resolve('Entity', 'b'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', 'attr'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', 'a'), ['attr']) + self.assertEqual(self.s.resolve_field('Entity', 'b'), ['attr']) def test_prefix(self): - self.assertEqual(self.s.resolve('Entity', 'sg_type'), ['sg_type']) - self.assertEqual(self.s.resolve('Entity', 'type'), ['sg_type']) - self.assertEqual(self.s.resolve('Entity', '!type'), ['type']) + self.assertEqual(self.s.resolve_field('Entity', 'sg_type'), ['sg_type']) + self.assertEqual(self.s.resolve_field('Entity', 'type'), ['sg_type']) + self.assertEqual(self.s.resolve_field('Entity', '!type'), ['type']) - self.assertEqual(self.s.resolve('Entity', 'sg_name'), ['sg_name']) - self.assertEqual(self.s.resolve('Entity', 'name'), ['name']) # different! - self.assertEqual(self.s.resolve('Entity', '!name'), ['name']) + self.assertEqual(self.s.resolve_field('Entity', 'sg_name'), ['sg_name']) + self.assertEqual(self.s.resolve_field('Entity', 'name'), ['name']) # different! + self.assertEqual(self.s.resolve_field('Entity', '!name'), ['name']) def test_missing_entity(self): - self.assertRaises(ValueError, self.s.resolve, 'Missing', 'field_name') + self.assertRaises(ValueError, self.s.resolve_field, 'Missing', 'field_name') def test_missing(self): - self.assertEqual(self.s.resolve('Entity', '$missing'), []) - self.assertEqual(self.s.resolve('Entity', '#missing'), []) - self.assertEqual(self.s.resolve('Entity', '!missing'), ['missing']) - self.assertEqual(self.s.resolve('Entity', 'missing'), ['missing']) - self.assertRaises(ValueError, self.s.resolve, 'Entity', 'missing', strict=True) + self.assertEqual(self.s.resolve_field('Entity', '$missing'), []) + self.assertEqual(self.s.resolve_field('Entity', '#missing'), []) + self.assertEqual(self.s.resolve_field('Entity', '!missing'), ['missing']) + self.assertEqual(self.s.resolve_field('Entity', 'missing'), ['missing']) + self.assertRaises(ValueError, self.s.resolve_field, 'Entity', 'missing', strict=True) + + + +class TestResolveDeepFields(TestCase): + + def setUp(self): + self.s = s = load_schema() + self.s.load({ + 'Task': { + 'field_aliases': { + 'status': 'sg_status_list', + 'parent': 'entity', + }, + 'field_tags': { + 'core': ['content', 'step', 'sg_status_list'], + } + }, + 'Shot': { + 'field_aliases': { + 'status': 'sg_status_list', + }, + 'field_tags': { + 'core': ['code', 'description', 'sg_sequence'], + } + }, + + 'Asset': { + 'field_aliases': { + 'status': 'sg_status_list', + }, + 'field_tags': { + 'core': ['code', 'asset_type'], + } + }, + }) + + def test_sanity(self): + self.assertEqual(self.s.resolve_field('Task', 'sg_status_list'), ['sg_status_list']) + self.assertEqual(self.s.resolve_field('Task', 'status_list'), ['sg_status_list']) + self.assertEqual(self.s.resolve_field('Task', '$status'), ['sg_status_list']) + + def test_passthrough(self): + self.assertEqual(self.s.resolve_field('Task', 'entity.Shot.sg_status_list'), ['entity.Shot.sg_status_list']) + + def test_explicit_aliases(self): + self.assertEqual(self.s.resolve_field('Task', '$parent.Shot.$status'), ['entity.Shot.sg_status_list']) + self.assertEqual(self.s.resolve_field('Task', '$parent.Shot.status_list'), ['entity.Shot.sg_status_list']) + self.assertEqual(self.s.resolve_field('Task', 'entity.Shot.$status'), ['entity.Shot.sg_status_list']) + + def test_explicit_tags(self): + self.assertEqual(self.s.resolve_field('Task', '$parent.Shot.#core'), ['entity.Shot.sg_sequence', 'entity.Shot.code', 'entity.Shot.description']) + +