Source code for luna_handlers.classes.schemas.conditional_tags_policy

"""
Module contains schemas for conditional tags policy
"""
from typing import List

from classes.event import HandlerEvent as Event
from classes.schemas import types
from classes.schemas.base_schema import BaseSchema
from classes.schemas.filters import ComplexFilter


[docs] class Tag(types.CustomString): """Tag schema""" # tags max length max_length = types.MAX_TAG_LENGTH @classmethod def __get_pydantic_json_schema__(cls, field_schema): """Override default string validation to check tag pattern matching""" cls.__get_pydantic_json_schema__(field_schema) field_schema.update(pattern=r"^[a-zA-Z0-9_-]+$")
[docs] class ConditionalTagsPolicy(BaseSchema): """Conditional tags policy schema""" # policy tag tag: Tag # policy filters filters: ComplexFilter = ComplexFilter()
[docs] def execute(self, events: List[Event]) -> None: """ Execute additional tags policy Args: events: processing events """ for event in events: if self.filters.isEmpty or self.filters.isEventSatisfies(event): if self.tag not in event.raw["tags"]: event.raw["tags"].append(self.tag)