Skip to content

matchers

Attributes

Classes

ValueMatcher

Bases: KiaraModel

An object describing requirements values should satisfy in order to be included in a query result.

Source code in src/kiara/models/values/matchers.py 
 14
 15
 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
class ValueMatcher(KiaraModel):
    """An object describing requirements values should satisfy in order to be included in a query result."""

    @classmethod
    def create_matcher(self, **match_options: Any):
        m = ValueMatcher(**match_options)
        return m

    data_types: List[str] = Field(description="The data type.", default_factory=list)
    allow_sub_types: bool = Field(description="Allow subtypes.", default=True)
    min_size: int = Field(description="The minimum size for the dataset.", default=0)
    max_size: Union[None, int] = Field(
        description="The maximum size for the dataset.", default=None
    )
    allow_internal: bool = Field(
        description="Allow internal data types.", default=False
    )
    has_alias: bool = Field(
        description="Value must have at least one alias.", default=True
    )
    alias_matchers: Union[None, List[str]] = Field(
        description="Values must have an alias that matches one of the provided matchers. Assumes 'has_alias' is set to 'True'.",
        default=None,
    )
    in_data_archives: Union[None, List[str]] = Field(
        description="A list of registered names of archives the value must be in. If 'None', all archives will be used.",
        default=None,
    )

    @field_validator("in_data_archives", mode="before")
    @classmethod
    def validate_in_archives(cls, v):
        if v is None:
            return v
        elif isinstance(v, str):
            v = set(v)
        else:
            v = set(v)

        if DEFAULT_STORE_MARKER in v or DEFAULT_DATA_STORE_MARKER in v:
            v.add(DEFAULT_DATA_STORE_MARKER)
            v.add(DEFAULT_STORE_MARKER)

        return list(v)

    @field_validator("alias_matchers")
    @classmethod
    def validate_matchers(cls, v):
        if v is None:
            return v
        elif isinstance(v, str):
            return [v]
        else:
            return list(v)

    def is_match(self, value: Value, kiara: "Kiara") -> bool:
        has_alias = self.has_alias or self.alias_matchers

        match = False
        if self.data_types:
            if not self.allow_sub_types:
                for data_type in self.data_types:
                    if data_type == value.data_type_name:
                        match = True
                        break
            else:
                if value.data_type_name not in kiara.type_registry.get_data_type_names(
                    include_profiles=True
                ):
                    return False
                lineage = kiara.type_registry.get_type_lineage(value.data_type_name)
                for data_type in self.data_types:
                    if data_type in lineage:
                        match = True
                        break
            if not match:
                return False

        if self.min_size:
            if value.value_size < self.min_size:
                return False
        if self.max_size:
            if value.value_size > self.max_size:
                return False

        if not self.allow_internal:
            if kiara.type_registry.is_internal_type(
                data_type_name=value.data_type_name
            ):
                return False

        if has_alias:
            aliases = kiara.alias_registry.find_aliases_for_value_id(
                value_id=value.value_id
            )
            if not aliases:
                return False

            if self.alias_matchers:
                for token in self.alias_matchers:
                    for alias in aliases:
                        if token in alias:
                            return True

                return False

        return True

Attributes

data_types: List[str] = Field(description='The data type.', default_factory=list) class-attribute instance-attribute
allow_sub_types: bool = Field(description='Allow subtypes.', default=True) class-attribute instance-attribute
min_size: int = Field(description='The minimum size for the dataset.', default=0) class-attribute instance-attribute
max_size: Union[None, int] = Field(description='The maximum size for the dataset.', default=None) class-attribute instance-attribute
allow_internal: bool = Field(description='Allow internal data types.', default=False) class-attribute instance-attribute
has_alias: bool = Field(description='Value must have at least one alias.', default=True) class-attribute instance-attribute
alias_matchers: Union[None, List[str]] = Field(description="Values must have an alias that matches one of the provided matchers. Assumes 'has_alias' is set to 'True'.", default=None) class-attribute instance-attribute
in_data_archives: Union[None, List[str]] = Field(description="A list of registered names of archives the value must be in. If 'None', all archives will be used.", default=None) class-attribute instance-attribute

Functions

create_matcher(**match_options: Any) classmethod
Source code in src/kiara/models/values/matchers.py 
17
18
19
20
@classmethod
def create_matcher(self, **match_options: Any):
    m = ValueMatcher(**match_options)
    return m
validate_in_archives(v) classmethod
Source code in src/kiara/models/values/matchers.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@field_validator("in_data_archives", mode="before")
@classmethod
def validate_in_archives(cls, v):
    if v is None:
        return v
    elif isinstance(v, str):
        v = set(v)
    else:
        v = set(v)

    if DEFAULT_STORE_MARKER in v or DEFAULT_DATA_STORE_MARKER in v:
        v.add(DEFAULT_DATA_STORE_MARKER)
        v.add(DEFAULT_STORE_MARKER)

    return list(v)
validate_matchers(v) classmethod
Source code in src/kiara/models/values/matchers.py 
59
60
61
62
63
64
65
66
67
@field_validator("alias_matchers")
@classmethod
def validate_matchers(cls, v):
    if v is None:
        return v
    elif isinstance(v, str):
        return [v]
    else:
        return list(v)
is_match(value: Value, kiara: Kiara) -> bool
Source code in src/kiara/models/values/matchers.py 
 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
def is_match(self, value: Value, kiara: "Kiara") -> bool:
    has_alias = self.has_alias or self.alias_matchers

    match = False
    if self.data_types:
        if not self.allow_sub_types:
            for data_type in self.data_types:
                if data_type == value.data_type_name:
                    match = True
                    break
        else:
            if value.data_type_name not in kiara.type_registry.get_data_type_names(
                include_profiles=True
            ):
                return False
            lineage = kiara.type_registry.get_type_lineage(value.data_type_name)
            for data_type in self.data_types:
                if data_type in lineage:
                    match = True
                    break
        if not match:
            return False

    if self.min_size:
        if value.value_size < self.min_size:
            return False
    if self.max_size:
        if value.value_size > self.max_size:
            return False

    if not self.allow_internal:
        if kiara.type_registry.is_internal_type(
            data_type_name=value.data_type_name
        ):
            return False

    if has_alias:
        aliases = kiara.alias_registry.find_aliases_for_value_id(
            value_id=value.value_id
        )
        if not aliases:
            return False

        if self.alias_matchers:
            for token in self.alias_matchers:
                for alias in aliases:
                    if token in alias:
                        return True

            return False

    return True