"""SchemaSectionBase class."""
import re
from typing import TYPE_CHECKING, Any, Dict, Generic, List, Type, TypeVar
from text_lint.exceptions.schema import (
LookupExpressionInvalid,
LookupExpressionInvalidDuplicatePositional,
LookupExpressionInvalidSequence,
)
from text_lint.utilities.translations import _, f
if TYPE_CHECKING: # pragma: no cover
from text_lint.operations.bases.operation_base import OperationBase
from text_lint.schema import Schema
AliasYamlOperation = Dict[str, Any]
TypeOperation = TypeVar("TypeOperation", bound="OperationBase[Any]")
[Doku]
class SchemaSectionBase(Generic[TypeOperation]):
"""A section of the schema file containing operation definitions."""
operation_classes: Dict[str, Type["TypeOperation"]]
entity_name: "str"
msg_fmt_invalid_regex = _("{0} #{1} Invalid regex")
msg_fmt_invalid_lookup_expression = _(
"{0} #{1} Invalid lookup expression: '{2}'"
)
msg_fmt_invalid_lookup_expression_duplicate_positional = _(
"{0} #{1} Invalid lookup expression: duplicate use of '{2}'"
)
msg_fmt_invalid_lookup_expression_sequence = _(
"{0} #{1} Transformation lookup in wrong sequence: '{2}'"
)
msg_fmt_restricted_operation = _(
"{0} #{1} This operation is for internal use only"
)
msg_fmt_unknown_operation = _("{0} #{1} Unknown operation")
msg_fmt_unknown_syntax = _("{0} #{1} Unknown syntax")
[Doku]
def __init__(self, schema: "Schema") -> None:
"""Initialize SchemaSectionBase instances.
:param schema: The schema instance this section is found in.
"""
self._schema = schema
[Doku]
def load(
self,
source: List[AliasYamlOperation],
) -> List["TypeOperation"]:
"""Load the operation instances from the schema.
:param source: The YAML source that defines the section's operations.
:returns: The new operation instances.
:raises: SchemaError
"""
operation_instances: List["TypeOperation"] = []
operation_definitions: List["AliasYamlOperation"] = []
for operation_index, operation_definition in enumerate(source):
operation_instance = self._parse_schema_instances(
operation_definition,
operation_index,
)
operation_definitions.append(operation_definition)
operation_instances.append(operation_instance)
operation_instances = self.hook_load_operation_instances(
operation_instances,
operation_definitions,
)
return operation_instances
def _parse_schema_instances(
self,
operation_definition: Dict[str, Any],
operation_index: int,
) -> "TypeOperation":
try:
linter_operation = operation_definition["operation"]
try:
operation_class = self.operation_classes[linter_operation]
if operation_class.internal_use_only:
raise self._schema.create_exception(
description=f(
self.msg_fmt_restricted_operation,
self.entity_name,
operation_index + 1,
nl=1,
),
operation_definition=operation_definition,
)
operation_definition = self.hook_create_operation_instance(
operation_class,
operation_definition,
)
except KeyError as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_unknown_operation,
self.entity_name,
operation_index + 1,
nl=1,
),
operation_definition=operation_definition,
) from exc
del operation_definition["operation"]
return operation_class(**operation_definition)
except (AttributeError, TypeError, KeyError) as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_unknown_syntax,
self.entity_name,
operation_index + 1,
nl=1,
),
operation_definition=operation_definition,
) from exc
except LookupExpressionInvalid as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_invalid_lookup_expression,
self.entity_name,
operation_index + 1,
exc.args[0],
nl=1,
),
operation_definition=operation_definition,
) from exc
except LookupExpressionInvalidDuplicatePositional as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_invalid_lookup_expression_duplicate_positional,
self.entity_name,
operation_index + 1,
exc.args[0],
nl=1,
),
operation_definition=operation_definition,
) from exc
except LookupExpressionInvalidSequence as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_invalid_lookup_expression_sequence,
self.entity_name,
operation_index + 1,
exc.args[0],
nl=1,
),
operation_definition=operation_definition,
) from exc
except re.error as exc:
raise self._schema.create_exception(
description=f(
self.msg_fmt_invalid_regex,
self.entity_name,
operation_index + 1,
nl=1,
),
operation_definition=operation_definition,
) from exc
[Doku]
def hook_load_operation_instances(
self,
operation_instances: List["TypeOperation"],
# pylint: disable=unused-argument
operation_definitions: List["AliasYamlOperation"],
) -> List["TypeOperation"]:
"""Modify the operation instances prior to returning loaded results.
:param operation_instances: The newly created operation instances.
:param operation_definitions: The YAML config for the operations.
:returns: The modified operation instances.
"""
return operation_instances
# pylint: disable=unused-argument
[Doku]
def hook_create_operation_instance(
self,
operation_class: Type["TypeOperation"],
yaml_definition: "AliasYamlOperation",
) -> "AliasYamlOperation":
"""Modify the yaml definition prior to creating the operation instance.
:param operation_class: The operation class being created.
:param yaml_definition: The YAML config for this operation.
:returns: The modified YAML config.
"""
return yaml_definition