schema.py 66.7 KB
Newer Older
1
# copyright 2004-2014 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
#
# This file is part of yams.
#
# yams is free software: you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
#
# yams is distributed in the hope that it will be useful, but WITHOUT ANY
# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with yams. If not, see <http://www.gnu.org/licenses/>.
Laurent Peuch's avatar
Laurent Peuch committed
18

Sylvain Thénault's avatar
Sylvain Thénault committed
19
"""Classes to define generic Entities/Relations schemas."""
root's avatar
root committed
20

Laurent Peuch's avatar
Laurent Peuch committed
21
22
23
from logilab.common.logging_ext import set_log_methods

import logging
root's avatar
root committed
24

25
import warnings
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
26
from copy import deepcopy
27
from itertools import chain
28
29
30
31
32
33
34
35
36
37
38
39
40
from typing import (
    Dict,
    Any,
    Type,
    TYPE_CHECKING,
    Sequence,
    Generator,
    List,
    Optional,
    Set,
    Tuple,
    Union,
)
41

42
from logilab.common.decorators import cached, clear_cache
root's avatar
root committed
43
from logilab.common.interface import implements
44
from logilab.common import deprecation
root's avatar
root committed
45

46
import yams
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
from yams import (
    BASE_TYPES,
    MARKER,
    ValidationError,
    BadSchemaDefinition,
    KNOWN_METAATTRIBUTES,
    convert_default_value,
    DEFAULT_ATTRPERMS,
    DEFAULT_COMPUTED_RELPERMS,
)
from yams.interfaces import (
    ISchema,
    IRelationSchema,
    IEntitySchema,
    IConstraint,
    IVocabularyConstraint,
)
Laurent Peuch's avatar
Laurent Peuch committed
64
from yams.constraints import BASE_CHECKERS, BASE_CONVERTERS, UniqueConstraint, BaseConstraint
65
import yams.types as yams_types
66

Laurent Peuch's avatar
Laurent Peuch committed
67
__docformat__: str = "restructuredtext en"
68

Laurent Peuch's avatar
Laurent Peuch committed
69
70
71
_: Type[str] = str


72
73
@deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
def role_name(relation_type, role) -> str:
Sylvain Thénault's avatar
cleanup    
Sylvain Thénault committed
74
75
76
    """function to use for qualifying attribute / relation in ValidationError
    errors'dictionnary
    """
77
    return "%s-%s" % (relation_type, role)
78

79

Laurent Peuch's avatar
Laurent Peuch committed
80
def rehash(dictionary: Dict) -> dict:
81
82
83
84
85
86
87
88
89
90
91
    """this function manually builds a copy of `dictionary` but forces
    hash values to be recomputed. Note that dict(d) or d.copy() don't
    do that.

    It is used to :
      - circumvent Pyro / (un)pickle problems (hash mode is changed
        during reconstruction)
      - force to recompute keys' hash values. This is needed when a
        schema's type is changed because the schema's hash method is based
        on the type attribute. This problem is illusrated by the pseudo-code
        below :
92

93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
        >>> topic = EntitySchema(type='Topic')
        >>> d = {topic : 'foo'}
        >>> d[topic]
        'foo'
        >>> d['Topic']
        'foo'
        >>> topic.type = 'folder'
        >>> topic in d
        False
        >>> 'Folder' in d
        False
        >>> 'Folder' in d.keys() # but it can be found "manually"
        True
        >>> d = rehash(d) # explicit rehash()
        >>> 'Folder' in d
        True
    """
    return dict(item for item in dictionary.items())
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
111

112

Nicolas Chauvat's avatar
Nicolas Chauvat committed
113
class ERSchema:
114
    """Base class shared by entity and relation schema."""
Nicolas Chauvat's avatar
Nicolas Chauvat committed
115

116
    @deprecation.argument_renamed(old_name="erdef", new_name="entity_relation_definition")
117
118
119
    def __init__(
        self, schema: "Schema", entity_relation_definition: yams_types.RelationDefinition = None
    ) -> None:
120
121
122
123
        """
        Construct an ERSchema instance.

        :Parameters:
124
         - `schema`: (??)
125
         - `entity_relation_definition`: (??)
126
        """
127
        if entity_relation_definition is None:
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
128
            return
Laurent Peuch's avatar
Laurent Peuch committed
129

"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
130
        self.schema = schema
131
        self.type: yams_types.DefinitionName = entity_relation_definition.name
132
133
        self.description: str = entity_relation_definition.description or ""
        self.package = entity_relation_definition.package
root's avatar
root committed
134

Laurent Peuch's avatar
Laurent Peuch committed
135
    def __eq__(self, other) -> bool:
136
        return self.type == getattr(other, "type", other)
137

Laurent Peuch's avatar
Laurent Peuch committed
138
    def __ne__(self, other) -> bool:
139
140
        return not (self == other)

Laurent Peuch's avatar
Laurent Peuch committed
141
    def __lt__(self, other) -> bool:
142
        return self.type < getattr(other, "type", other)
143

Laurent Peuch's avatar
Laurent Peuch committed
144
    def __hash__(self) -> int:
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
145
        try:
146
            return hash(self.type)
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
147
148
149
        except AttributeError:
            pass
        return hash(id(self))
150

151
    def __deepcopy__(self: "ERSchema", memo) -> "ERSchema":
152
        clone = self.__class__(deepcopy(self.schema, memo))
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
153
154
155
        memo[id(self)] = clone
        clone.type = deepcopy(self.type, memo)
        clone.__dict__ = deepcopy(self.__dict__, memo)
Laurent Peuch's avatar
Laurent Peuch committed
156

"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
157
        return clone
158

Laurent Peuch's avatar
Laurent Peuch committed
159
    def __str__(self) -> str:
160
        return self.type
161

root's avatar
root committed
162

163
class PermissionMixIn:
164
    """mixin class for permissions handling"""
165

Laurent Peuch's avatar
Laurent Peuch committed
166
167
    # https://github.com/python/mypy/issues/5837
    if TYPE_CHECKING:
168
169

        def __init__(
170
            self,
171
172
            schema: Optional["Schema"],
            relation_definition: Optional[yams_types.EntityType],
173
174
            *args,
            **kwargs,
175
        ):
176
            self.permissions: yams_types.Permissions
177

Laurent Peuch's avatar
Laurent Peuch committed
178
            # fake init for mypy
179
180
181
182
            if relation_definition:
                self.permissions = relation_definition.__permissions__.copy()
            else:
                self.permissions = {}
Laurent Peuch's avatar
Laurent Peuch committed
183
184
185
186
            self.final: bool = True

        @property
        def ACTIONS(self) -> Tuple[str, ...]:
187
            return ("read", "add", "update", "delete")
Laurent Peuch's avatar
Laurent Peuch committed
188
189
190
191

        def advertise_new_add_permission(self) -> None:
            pass

192
    def action_permissions(self, action: str) -> yams_types.Permission:
193
        return self.permissions[action]
194

195
    def set_action_permissions(self, action: str, permissions: yams_types.Permission) -> None:
196
197
198
        assert type(permissions) is tuple, "permissions is expected to be a tuple not %s" % type(
            permissions
        )
Laurent Peuch's avatar
Laurent Peuch committed
199

200
        assert action in self.ACTIONS, "%s not in %s" % (action, self.ACTIONS)
Laurent Peuch's avatar
Laurent Peuch committed
201

202
203
        self.permissions[action] = permissions

Laurent Peuch's avatar
Laurent Peuch committed
204
    def check_permission_definitions(self) -> None:
205
        """check permissions are correctly defined"""
Laurent Peuch's avatar
Laurent Peuch committed
206

207
208
        # already initialized, check everything is fine
        for action, groups in self.permissions.items():
209
            assert action in self.ACTIONS, "unknown action %s for %s" % (action, self)
Laurent Peuch's avatar
Laurent Peuch committed
210

211
212
213
            assert isinstance(
                groups, tuple
            ), "permission for action %s of %s isn't a tuple as " "expected" % (action, self)
Laurent Peuch's avatar
Laurent Peuch committed
214

215
216
        if self.final:
            self.advertise_new_add_permission()
Laurent Peuch's avatar
Laurent Peuch committed
217

218
        for action in self.ACTIONS:
219
220
221
            assert (
                action in self.permissions
            ), "missing expected permissions for action %s for %s" % (action, self)
222

root's avatar
root committed
223
224
225

# Schema objects definition ###################################################

226

227
@deprecation.attribute_renamed(old_name="subjrels", new_name="_subject_relations")
228
@deprecation.attribute_renamed(old_name="objrels", new_name="_object_relations")
229
class EntitySchema(PermissionMixIn, ERSchema):
230
    """An entity has a type, a set of subject and or object relations
root's avatar
root committed
231
    the entity schema defines the possible relations for a given type and some
232
    constraints on those relations.
root's avatar
root committed
233
    """
234

235
236
    __implements__ = IEntitySchema

237
    ACTIONS: Tuple[str, ...] = ("read", "add", "update", "delete")
238
239
    field_checkers: yams_types.Checkers = BASE_CHECKERS
    field_converters: yams_types.Converters = BASE_CONVERTERS
Nicolas Chauvat's avatar
Nicolas Chauvat committed
240

241
242
    # XXX set default values for those attributes on the class level since
    # they may be missing from schemas obtained by pyro
Laurent Peuch's avatar
Laurent Peuch committed
243
244
    _specialized_type: Optional[str] = None
    _specialized_by: List = []
Sylvain Thénault's avatar
cleanup    
Sylvain Thénault committed
245

246
    @deprecation.argument_renamed(old_name="rdef", new_name="relation_definition")
247
    def __init__(
248
        self,
249
250
        schema: "Schema" = None,
        relation_definition: yams_types.EntityType = None,
251
252
        *args,
        **kwargs,
253
    ) -> None:
254
        super(EntitySchema, self).__init__(schema, relation_definition, *args, **kwargs)
Laurent Peuch's avatar
Laurent Peuch committed
255

256
        if relation_definition is not None:
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
257
            # quick access to bounded relation schemas
258
259
            self._subject_relations: Dict[Union[str, "RelationSchema"], "RelationSchema"] = {}
            self._object_relations: Dict[Union[str, "RelationSchema"], "RelationSchema"] = {}
260
261
            self._specialized_type = relation_definition.specialized_type
            self._specialized_by = relation_definition.specialized_by
Laurent Peuch's avatar
Laurent Peuch committed
262
            self.final: bool = self.type in BASE_TYPES
263
264
265
266
            self.permissions: Dict[
                str, Tuple[str, ...]
            ] = relation_definition.__permissions__.copy()
            self._unique_together: List = getattr(relation_definition, "__unique_together__", [])
Laurent Peuch's avatar
Laurent Peuch committed
267

268
        else:  # this happens during deep copy (cf. ERSchema.__deepcopy__)
269
270
            self._specialized_type = None
            self._specialized_by = []
271

Laurent Peuch's avatar
Laurent Peuch committed
272
    def check_unique_together(self) -> None:
273
        errors = []
Laurent Peuch's avatar
Laurent Peuch committed
274

275
276
277
        for unique_together in self._unique_together:
            for name in unique_together:
                try:
278
                    relation_schema = self.relation_definition(name, take_first=True)
279
                except KeyError:
280
                    errors.append("no such attribute or relation %s" % name)
281
                else:
282
                    if not (relation_schema.final or relation_schema.relation_type.inlined):
283
                        errors.append("%s is not an attribute or an inlined " "relation" % name)
Laurent Peuch's avatar
Laurent Peuch committed
284

285
        if errors:
286
287
288
289
290
            message = "invalid __unique_together__ specification for %s: %s" % (
                self,
                ", ".join(errors),
            )
            raise BadSchemaDefinition(message)
291

Laurent Peuch's avatar
Laurent Peuch committed
292
    def __repr__(self) -> str:
293
294
        return "<%s %s - %s>" % (
            self.type,
295
296
            [subject_relation.type for subject_relation in self.subject_relations()],
            [object_relation.type for object_relation in self.object_relations()],
297
        )
298

Laurent Peuch's avatar
Laurent Peuch committed
299
    def _rehash(self) -> None:
300
        self._subject_relations = rehash(self._subject_relations)
301
        self._object_relations = rehash(self._object_relations)
302

Laurent Peuch's avatar
Laurent Peuch committed
303
    def advertise_new_add_permission(self) -> None:
304
305
        pass

root's avatar
root committed
306
    # schema building methods #################################################
307

308
    @deprecation.argument_renamed(old_name="rschema", new_name="relation_schema")
309
    def add_subject_relation(self, relation_schema: "RelationSchema") -> None:
root's avatar
root committed
310
        """register the relation schema as possible subject relation"""
311
        self._subject_relations[relation_schema] = relation_schema
Laurent Peuch's avatar
Laurent Peuch committed
312

313
314
        clear_cache(self, "ordered_relations")
        clear_cache(self, "meta_attributes")
315

316
    @deprecation.argument_renamed(old_name="rschema", new_name="relation_schema")
317
    def add_object_relation(self, relation_schema: "RelationSchema") -> None:
root's avatar
root committed
318
        """register the relation schema as possible object relation"""
319
        self._object_relations[relation_schema] = relation_schema
320

321
322
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_schema")
    def del_subject_relation(self, relation_schema: "RelationSchema") -> None:
323
        try:
324
            del self._subject_relations[relation_schema]
325
326
            clear_cache(self, "ordered_relations")
            clear_cache(self, "meta_attributes")
327
        except KeyError:
328
            pass  # XXX error should never pass silently
329

330
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
331
332
333
    def del_object_relation(self, relation_schema: "RelationSchema") -> None:
        if relation_schema in self._object_relations:
            del self._object_relations[relation_schema]
root's avatar
root committed
334
335
336

    # IEntitySchema interface #################################################

337
    # navigation ######################
338

339
    def specializes(self) -> Optional["EntitySchema"]:
Laurent Peuch's avatar
Laurent Peuch committed
340
        if self._specialized_type and self.schema is not None:
341
            return self.schema.entity_schema_for(yams_types.DefinitionName(self._specialized_type))
342
343
        return None

344
    def ancestors(self) -> List["EntitySchema"]:
345
346
        specializes = self.specializes()
        ancestors = []
Laurent Peuch's avatar
Laurent Peuch committed
347

348
349
350
        while specializes:
            ancestors.append(specializes)
            specializes = specializes.specializes()
Laurent Peuch's avatar
Laurent Peuch committed
351

352
353
        return ancestors

354
    def specialized_by(self, recursive: bool = True) -> List["EntitySchema"]:
Laurent Peuch's avatar
Laurent Peuch committed
355
356
357
        if not self.schema:
            return []

358
        entity_schema = self.schema.entity_schema_for
359
        subject_schemas = [entity_schema(entity_type) for entity_type in self._specialized_by]
Laurent Peuch's avatar
Laurent Peuch committed
360

361
        if recursive:
362
363
            for subject_schema in subject_schemas[:]:
                subject_schemas.extend(subject_schema.specialized_by(recursive=True))
Laurent Peuch's avatar
Laurent Peuch committed
364

365
        return subject_schemas
366

367
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
368
    def has_relation(self, relation_type: yams_types.RelationType, role: str) -> bool:
369
        if role == "subject":
370
            return relation_type in self._subject_relations
Laurent Peuch's avatar
Laurent Peuch committed
371

372
        return relation_type in self._object_relations
373

374
    def subject_relations(self) -> List["RelationSchema"]:
root's avatar
root committed
375
376
377
        """return a list of relations that may have this type of entity as
        subject
        """
378
        return list(self._subject_relations.values())
379

380
    def object_relations(self) -> List["RelationSchema"]:
root's avatar
root committed
381
382
383
        """return a list of relations that may have this type of entity as
        object
        """
384
        return list(self._object_relations.values())
root's avatar
root committed
385

386
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
387
    @deprecation.argument_renamed(old_name="targettype", new_name="target_type")
388
    @deprecation.argument_renamed(old_name="takefirst", new_name="take_first")
389
    def relation_definition(
390
        self,
391
        relation_type: yams_types.DefinitionName,
392
        role: str = "subject",
393
        target_type: yams_types.DefinitionName = None,
394
        take_first: bool = False,
395
396
    ) -> yams_types.RelationDefinitionSchema:

397
        """return a relation definition schema for a relation of this entity type
398

399
        Notice that when target_type is not specified and the relation may lead
400
        to different entity types (ambiguous relation), one of them is picked
401
        randomly. If also take_first is False, a warning will be emitted.
402
        """
Laurent Peuch's avatar
Laurent Peuch committed
403
        assert self.schema is not None
404
        relation_schema = self.schema.relation_schema_for(relation_type)
Laurent Peuch's avatar
Laurent Peuch committed
405

406
        if target_type is None:
407
            if role == "subject":
408
                types = relation_schema.objects(self)
409
            else:
410
                types = relation_schema.subjects(self)
Laurent Peuch's avatar
Laurent Peuch committed
411

412
            if len(types) != 1 and not take_first:
413
                warnings.warn(
414
                    "[yams 0.38] no target_type specified and there are several "
415
                    "relation definitions for relation_type %s: %s. Yet you get the first "
416
                    "relation_definition."
417
                    % (relation_type, [entity_schema.type for entity_schema in types]),
418
419
420
                    Warning,
                    stacklevel=2,
                )
Laurent Peuch's avatar
Laurent Peuch committed
421

422
            target_type = types[0].type
Laurent Peuch's avatar
Laurent Peuch committed
423

424
        return relation_schema.role_relation_definition(self.type, target_type, role)
425
426

    rdef = deprecation.renamed(old_name="rdef", new_function=relation_definition)
427

428
    @cached
429
    def ordered_relations(self) -> List["RelationSchema"]:
430
        """return subject relations in an ordered way"""
Laurent Peuch's avatar
Laurent Peuch committed
431
432
        # mypy: "RelationDefinitionSchema" has no attribute "order"
        # this is a dynamically setted attribue using self.__dict__.update(some_dict)
433
        return sorted(
434
            self._subject_relations.values(),
435
            key=lambda x: x.relation_definition(self, x.objects(self)[0]).order,  # type: ignore
436
        )
Laurent Peuch's avatar
Laurent Peuch committed
437

438
    _RelationDefinitionsReturnType = Generator[
439
        Tuple["RelationSchema", Tuple["EntitySchema", ...], str], Any, None
440
    ]
441

442
443
    @deprecation.argument_renamed(old_name="includefinal", new_name="include_final")
    def relation_definitions(self, include_final: bool = False) -> "_RelationDefinitionsReturnType":
444
445
        """return an iterator on relation definitions

446
        if include_final is false, only non attribute relation are returned
447
448
449
450
451

        a relation definition is a 3-uple :
        * schema of the (non final) relation
        * schemas of the possible destination entity types
        * a string telling if this is a 'subject' or 'object' relation
root's avatar
root committed
452
        """
453
        for relation_schema in self.ordered_relations():
454
            if include_final or not relation_schema.final:
455
                yield relation_schema, relation_schema.objects(self), "subject"
Laurent Peuch's avatar
Laurent Peuch committed
456

457
458
        for relation_schema in self.object_relations():
            yield relation_schema, relation_schema.subjects(self), "object"
root's avatar
root committed
459

460
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
461
    def destination(self, relation_type: Union[str, "RelationSchema"]) -> "EntitySchema":
462
        """return the type or schema of entities related by the given subject relation
463

464
        `relation_type` is expected to be a non ambiguous relation
root's avatar
root committed
465
        """
466
        relation_schema = self._subject_relations[relation_type]
467
        object_types = relation_schema.objects(self.type)
Laurent Peuch's avatar
Laurent Peuch committed
468

469
470
471
472
473
        assert len(object_types) == 1, (
            self.type,
            str(relation_type),
            [str(ot) for ot in object_types],
        )
Laurent Peuch's avatar
Laurent Peuch committed
474

475
        return object_types[0]
476

477
    # attributes description ###########
478

479
480
    def attribute_definitions(
        self,
481
    ) -> Generator[Tuple["RelationSchema", "EntitySchema"], Any, None]:
482
        """return an iterator on attribute definitions
483

484
485
        attribute relations are a subset of subject relations where the
        object's type is a final entity
486

487
488
489
490
        an attribute definition is a 2-uple :
        * schema of the (final) relation
        * schema of the destination entity type
        """
491
492
        for relation_schema in self.ordered_relations():
            if not relation_schema.final:
493
                continue
Laurent Peuch's avatar
Laurent Peuch committed
494

495
            yield relation_schema, relation_schema.objects(self)[0]
root's avatar
root committed
496

497
    def main_attribute(self) -> Optional["RelationSchema"]:
498
499
500
        """convenience method that returns the *main* (i.e. the first non meta)
        attribute defined in the entity schema
        """
501
502
503
        for relation_schema, _ in self.attribute_definitions():
            if not self.is_metadata(relation_schema):
                return relation_schema
504

Laurent Peuch's avatar
Laurent Peuch committed
505
506
        return None

507
    def defaults(self) -> Generator[Tuple["RelationSchema", Any], Any, None]:
508
        """return an iterator on (attribute name, default value)"""
509
510
511
        for relation_schema in self.subject_relations():
            if relation_schema.final:
                value = self.default(relation_schema.type)
Laurent Peuch's avatar
Laurent Peuch committed
512

513
                if value is not None:
514
                    yield relation_schema, value
515

516
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
517
    def default(self, relation_type: yams_types.DefinitionName) -> Any:
518
        """return the default value of a subject relation"""
519
        relation_definition = self.relation_definition(relation_type, take_first=True)
Laurent Peuch's avatar
Laurent Peuch committed
520
521
        # mypy: "RelationDefinitionSchema" has no attribute "default"
        # this is a dynamically setted attribue using self.__dict__.update(some_dict)
522
        default = relation_definition.default  # type: ignore
Laurent Peuch's avatar
Laurent Peuch committed
523

524
525
        if callable(default):
            default = default()
Laurent Peuch's avatar
Laurent Peuch committed
526

527
528
529
        if default is MARKER:
            default = None
        elif default is not None:
530
            return convert_default_value(relation_definition, default)
Laurent Peuch's avatar
Laurent Peuch committed
531

532
        return default
533

534
    @deprecation.argument_renamed(old_name="rtype", new_name="relation_type")
535
    def has_unique_values(self, relation_type: yams_types.DefinitionName) -> bool:
536
537
        """convenience method to check presence of the UniqueConstraint on a
        relation
538
        """
539
        return bool(self.relation_definition(relation_type).constraint_by_class(UniqueConstraint))
540

541
    # metadata attributes #############
542

543
    @cached
544
    def meta_attributes(self) -> Dict["RelationSchema", Tuple[str, str]]:
545
546
547
548
549
        """return a dictionnary defining meta-attributes:
        * key is an attribute schema
        * value is a 2-uple (metadata name, described attribute name)

        a metadata attribute is expected to be named using the following scheme:
550

551
552
553
554
555
          <described attribute name>_<metadata name>

        for instance content_format is the format metadata of the content
        attribute (if it exists).
        """
556
        meta_attributes = {}
Laurent Peuch's avatar
Laurent Peuch committed
557

558
        for relation_schema, _ in self.attribute_definitions():
559
            try:
560
                attribute, meta = relation_schema.type.rsplit("_", -1)
561
562
            except ValueError:
                continue
Laurent Peuch's avatar
Laurent Peuch committed
563

564
            if meta in KNOWN_METAATTRIBUTES and attribute in self._subject_relations:
565
                meta_attributes[relation_schema] = (meta, attribute)
Laurent Peuch's avatar
Laurent Peuch committed
566

567
        return meta_attributes
568

569
    @deprecation.argument_renamed(old_name="attr", new_name="attribute")
570
    def has_metadata(self, attribute, metadata) -> Optional["RelationSchema"]:
571
        """return metadata's relation schema if this entity has the given
572
        `metadata` field for the given `attribute` attribute
root's avatar
root committed
573
        """
574
        return self._subject_relations.get("%s_%s" % (attribute, metadata))
575

576
577
    @deprecation.argument_renamed(old_name="attr", new_name="attribute")
    def is_metadata(self, attribute) -> Optional[Tuple[str, str]]:
578
579
        """return a metadata for an attribute (None if unspecified)"""
        try:
580
            attribute, metadata = str(attribute).rsplit("_", 1)
581
582
        except ValueError:
            return None
Laurent Peuch's avatar
Laurent Peuch committed
583

584
        if metadata in KNOWN_METAATTRIBUTES and attribute in self._subject_relations:
585
            return (attribute, metadata)
Laurent Peuch's avatar
Laurent Peuch committed
586

587
588
589
        return None

    # full text indexation control #####
590

591
    def indexable_attributes(self) -> Generator["RelationSchema", Any, None]:
Adrien Di Mascio's avatar
Adrien Di Mascio committed
592
        """return the relation schema of attribtues to index"""
593
594
        for relation_schema in self.subject_relations():
            if relation_schema.final:
595
                try:
Laurent Peuch's avatar
Laurent Peuch committed
596
597
                    # mypy: "RelationDefinitionSchema" has no attribute "fulltextindexed"
                    # this is a dynamically setted attribue using self.__dict__.update(some_dict)
598
                    if self.relation_definition(relation_schema).fulltextindexed:  # type: ignore
599
                        yield relation_schema
600
                except AttributeError:
601
602
                    # fulltextindexed is only available on String / Bytes
                    continue
603

604
    def fulltext_relations(self) -> Generator[Tuple["RelationSchema", str], Any, None]:
605
        """return the (name, role) of relations to index"""
606
607
608
        for relation_schema in self.subject_relations():
            if not relation_schema.final and relation_schema.fulltext_container == "subject":
                yield relation_schema, "subject"
Laurent Peuch's avatar
Laurent Peuch committed
609

610
611
612
        for relation_schema in self.object_relations():
            if relation_schema.fulltext_container == "object":
                yield relation_schema, "object"
613

614
    def fulltext_containers(self) -> Generator[Tuple["RelationSchema", str], Any, None]:
615
616
617
        """return relations whose extremity points to an entity that should
        contains the full text index content of entities of this type
        """
618
619
620
        for relation_schema in self.subject_relations():
            if relation_schema.fulltext_container == "object":
                yield relation_schema, "object"
Laurent Peuch's avatar
Laurent Peuch committed
621

622
623
624
        for relation_schema in self.object_relations():
            if relation_schema.fulltext_container == "subject":
                yield relation_schema, "subject"
625

626
    # resource accessors ##############
627

628
    @deprecation.argument_renamed(old_name="skiprels", new_name="skip_relations")
629
630
631
    def is_subobject(
        self,
        strict: bool = False,
632
        skip_relations: Sequence[Tuple[Union["RelationSchema", str], str]] = (),
633
    ) -> bool:
634
635
        """return True if this entity type is contained by another. If strict,
        return True if this entity type *must* be contained by another.
636
        """
637
        for relation_schema in self.object_relations():
638
            if (relation_schema, "object") in skip_relations:
639
                continue
Laurent Peuch's avatar
Laurent Peuch committed
640

641
            relation_definition = self.relation_definition(
642
                relation_schema.type, "object", take_first=True
643
            )
Laurent Peuch's avatar
Laurent Peuch committed
644
645
646
647

            # mypy: "RelationDefinitionSchema" has no attribute "composite"
            # this is a dynamically setted attribue using self.__dict__.update(some_dict)
            # same for cardinality just after
648
649
            if relation_definition.composite == "subject":  # type: ignore
                if not strict or relation_definition.cardinality[1] in "1+":  # type: ignore
650
                    return True
Laurent Peuch's avatar
Laurent Peuch committed
651

652
        for relation_schema in self.subject_relations():
653
            if (relation_schema, "subject") in skip_relations:
654
                continue
Laurent Peuch's avatar
Laurent Peuch committed
655

656
            if relation_schema.final:
657
                continue
Laurent Peuch's avatar
Laurent Peuch committed
658

659
            relation_definition = self.relation_definition(
660
                relation_schema.type, "subject", take_first=True
661
            )
Laurent Peuch's avatar
Laurent Peuch committed
662
663
664
665

            # mypy: "RelationDefinitionSchema" has no attribute "composite"
            # this is a dynamically setted attribue using self.__dict__.update(some_dict)
            # same for cardinality just after
666
667
            if relation_definition.composite == "object":  # type: ignore
                if not strict or relation_definition.cardinality[0] in "1+":  # type: ignore
668
                    return True
Laurent Peuch's avatar
Laurent Peuch committed
669

670
        return False
671

672
    # validation ######################
673

674
675
    def check(
        self,
676
        entity: Dict[Union[str, "RelationSchema"], Any],
677
678
        creation: bool = False,
        _=None,
679
        relations: Optional[List["RelationSchema"]] = None,
680
    ) -> None:
"Sylvain ext:(%22)'s avatar
"Sylvain ext:(%22) committed
681
        """check the entity and raises an ValidationError exception if it
root's avatar
root committed
682
683
        contains some invalid fields (ie some constraints failed)
        """
Laurent Peuch's avatar
Laurent Peuch committed
684

685
        if _ is not None:
686
687
688
            warnings.warn(
                "[yams 0.36] _ argument is deprecated, remove it", DeprecationWarning, stacklevel=2
            )
Laurent Peuch's avatar
Laurent Peuch committed
689
690
691
692
693
694

        # mypy: Name '_' already defined on line 593
        # we force redeclaration because of the previous if
        # we probably want to remove all this very old depreciation code tbh...
        _: Type[str] = str  # type: ignore
        errors: Dict[str, str] = {}
695
        message_arguments: Dict[str, Any] = {}
Laurent Peuch's avatar
Laurent Peuch committed
696
        i18nvalues: List[str] = []
697
        relations = relations or self.subject_relations()
Laurent Peuch's avatar
Laurent Peuch committed
698

699
700
        for relation_schema in relations:
            if not relation_schema.final:
701
                continue
Laurent Peuch's avatar
Laurent Peuch committed
702

703
704
            aschema = self.destination(relation_schema)
            qname = role_name(relation_schema, "subject")
705
            relation_definition = relation_schema.relation_definition(self.type, aschema.type)
Laurent Peuch's avatar
Laurent Peuch committed
706

707
            # don't care about rhs cardinality, always '*' (if it make senses)
Laurent Peuch's avatar
Laurent Peuch committed
708
709
            # mypy: "RelationDefinitionSchema" has no attribute "cardinality"
            # this is a dynamically setted attribue using self.__dict__.update(some_dict)
710
            card = relation_definition.cardinality[0]  # type: ignore
Laurent Peuch's avatar
Laurent Peuch committed
711

712
            assert card in "?1"
Laurent Peuch's avatar
Laurent Peuch committed
713

714
            required = card == "1"
Laurent Peuch's avatar
Laurent Peuch committed
715

716
            # check value according to their type
717
718
            if relation_schema in entity:
                value = entity[relation_schema]
719
            else:
720
721
722
                if creation and required:
                    # missing required attribute with no default on creation
                    # is not autorized
723
                    errors[qname] = _("required attribute")
724
725
                # on edition, missing attribute is considered as no changes
                continue
Laurent Peuch's avatar
Laurent Peuch committed
726

727
            # skip other constraint if value is None and None is allowed
728
729
            if value is None:
                if required:
730
                    errors[qname] = _("required attribute")
Laurent Peuch's avatar
Laurent Peuch committed
731

732
                continue
Laurent Peuch's avatar
Laurent Peuch committed
733

734
            if not aschema.check_value(value):
735
                errors<