Source code for datamasque.client.models.status

import enum
from typing import Optional

from pydantic import BaseModel




class ValidationStatus(enum.Enum):
    """Validation status of a ruleset or ruleset library."""

    valid = "valid"
    invalid = "invalid"
    in_progress = "in_progress"
    unknown = "unknown"





class ValidationErrorType(enum.Enum):
    """Categorises why validation failed (see `ValidationErrorDetails.validation_error_type`)."""

    ruleset = "ruleset"
    library_missing = "library_missing"
    library_invalid = "library_invalid"
    expansion = "expansion"  # The ruleset is not valid once its library references are expanded.





class ValidationErrorDetails(BaseModel):
    """A single validation error, with its position in the YAML when available."""

    message: str
    validation_error_type: Optional[ValidationErrorType] = None
    line_number: Optional[int] = None
    column_number: Optional[int] = None





class MaskingRunStatus(enum.Enum):
    """List of valid masking run statuses."""

    finished = "finished"
    finished_with_warnings = "finished_with_warnings"
    queued = "queued"
    running = "running"
    failed = "failed"
    validating = "validating"
    cancelling = "cancelling"
    cancelled = "cancelled"



    @classmethod
    def get_final_states(cls) -> set["MaskingRunStatus"]:
        """Returns the list of final statuses, i.e. the run is completed, successfully or otherwise."""

        return {cls.finished, cls.finished_with_warnings, cls.cancelled, cls.failed}




    @classmethod
    def get_finished_states(cls) -> set["MaskingRunStatus"]:
        """Returns the list of statuses that indicate the run completed successfully."""

        return {cls.finished, cls.finished_with_warnings}


    @property
    def is_in_final_state(self) -> bool:
        """Returns True if this status is a final status."""

        return self in self.get_final_states()

    @property
    def is_finished(self) -> bool:
        """Returns True if this status is a finished status."""

        return self in self.get_finished_states()





class AsyncRulesetGenerationTaskStatus(enum.Enum):
    """List of statuses of async ruleset generation tasks."""

    finished = "finished"
    finished_with_warnings = "finished_with_warnings"
    failed = "failed"
    running = "running"
    queued = "queued"
    cancelled = "cancelled"



    @classmethod
    def get_final_states(cls) -> set["AsyncRulesetGenerationTaskStatus"]:
        """Returns the list of final statuses, i.e. the ruleset generation has completed, successfully or otherwise."""

        return {cls.finished, cls.finished_with_warnings, cls.failed, cls.cancelled}




    @classmethod
    def get_finished_states(cls) -> set["AsyncRulesetGenerationTaskStatus"]:
        """Returns the list of statuses that indicate the ruleset generation completed successfully."""

        return {cls.finished, cls.finished_with_warnings}


    @property
    def is_in_final_state(self) -> bool:
        """Returns True if this status is a final status."""

        return self in self.get_final_states()

    @property
    def is_finished(self) -> bool:
        """Returns True if this status is a finished status."""

        return self in self.get_finished_states()