dal.validation package

Submodules

dal.validation.constants module

Validation constants

dal.validation.flow_validator module

class dal.validation.flow_validator.FlowValidator(flow_ref: str)

Bases: object

Validates a specific flow within the project.

validate_flow() → ProjectValidationResult

Validate a specific flow and all subflows reachable from it.

Returns:

The result of the flow validation, including issues found.

Return type:

ProjectValidationResult

dal.validation.issues module

class dal.validation.issues.DuplicatedMob(json_path: Path, msg: str, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Duplicated flow or node instance in multiple packages

msg

Issue message.

Type:

str

class dal.validation.issues.Issue(category: str, iss_type: str, severity: Severity, msg: str, line_start: int | None = None, line_ranges: List[Dict[str, int]] | None = None)

Bases: object

Base issue class.

category

Issue category.

Type:

str

iss_type

Issue type.

Type:

str

severity

Issue severity.

Type:

Severity

msg

Issue message.

Type:

str

line_start

Issue start line.

Type:

int

line_ranges

Issue line range.

Type:

Any

class dal.validation.issues.MissingFlowInstance(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Missing flow instance referenced by link.

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.MissingMob(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Missing mob (node or flow).

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.MissingNodeInstance(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Missing node instance referenced by link.

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.MissingNodePort(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Missing node port.

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.MissingReferencedParameter(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Missing flow parameter referenced by node instance.

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.NonMatchingLinkPorts(json_path: Path, msg: str, line_start: int | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: ProjIssue

Link src and dst ports do not match (should not be connected).

json_path

Path to file with issue.

Type:

Path

msg

Issue message.

Type:

str

class dal.validation.issues.ProjIssue(category: str, iss_type: str, severity: Severity, msg: str, json_path: Path, line_start: int | None = None, line_ranges: List[Dict[str, int]] | None = None, document_type: str = 'Unknown', document_name: str = 'Unknown')

Bases: Issue

Project base issue class.

json_path

Path to file with issue.

Type:

Path

category

Issue category.

Type:

str

iss_type

Issue type.

Type:

str

severity

Issue severity.

Type:

Severity

msg

Issue message.

Type:

str

line_start

Issue start line.

Type:

int

line_ranges

Issue line range.

Type:

Any

class dal.validation.issues.Severity(value)

Bases: str, Enum

Severity enumeration.

ERROR = 'ERROR'
HIGH = 'HIGH'
LOW = 'LOW'
NORMAL = 'NORMAL'

dal.validation.project_validator module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

class dal.validation.project_validator.LinkValidator(objects_by_scope: Dict, node_dict_cache: Dict, logger)

Bases: object

Link validator class.

Validate a single link.

Parameters:

  • flow_ref – Name of the flow containing the link

  • flow_data – Full flow data dictionary (for line number lookup)

  • flow_content – Flow content dictionary

  • link_id – Link identifier

  • from_path – Link source path (format: “node__subnode/port/out”)

  • to_path – Link destination path (format: “node__subnode/port/in”)

exception dal.validation.project_validator.MissingFlowInstanceExc(flow: str, expected_instance: str)

Bases: Exception

Raised when a flow instance is not found.

flow

Flow where the instance is missing.

Type:

str

expected_instance

Instance missing.

Type:

str

exception dal.validation.project_validator.MissingFlowTemplateExc

Bases: Exception

Raised when a flow template is not found.

exception dal.validation.project_validator.MissingNodeInstanceExc(flow: str, expected_instance: str)

Bases: Exception

Raised when a node instance is not found.

flow

Flow where the instance is missing.

Type:

str

expected_instance

Instance missing.

Type:

str

exception dal.validation.project_validator.MissingNodeTemplateExc

Bases: Exception

Raised when a node template is not found.

class dal.validation.project_validator.ProjectIssue(*, category: str, iss_type: str, severity: Severity, msg: str, json_path: str, document_type: str, document_name: str, line_start: int | None = None)

Bases: BaseModel

Base class for project validation issues.

category: str
document_name: str
document_type: str
iss_type: str
json_path: str
line_start: int | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'category': FieldInfo(annotation=str, required=True), 'document_name': FieldInfo(annotation=str, required=True), 'document_type': FieldInfo(annotation=str, required=True), 'iss_type': FieldInfo(annotation=str, required=True), 'json_path': FieldInfo(annotation=str, required=True), 'line_start': FieldInfo(annotation=Union[int, NoneType], required=False), 'msg': FieldInfo(annotation=str, required=True), 'severity': FieldInfo(annotation=Severity, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

msg: str
severity: Severity
class dal.validation.project_validator.ProjectValidationResult(*, summary: Summary, issues: List[ProjectIssue])

Bases: BaseModel

Result structure for usage search.

Format: {

“summary”: {

“total_issues”: int, “errors_count”: int, “warnings_count”: int, “scopes_checked”: List[str]

}, “issues”: List[ProjectIssue]

}

issues: List[ProjectIssue]
model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'issues': FieldInfo(annotation=List[dal.validation.project_validator.ProjectIssue], required=True), 'summary': FieldInfo(annotation=Summary, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

summary: Summary
class dal.validation.project_validator.ProjectValidator

Bases: object

Validates all movai project data according to: - No duplicate MOB names across workspace - For each Flow, check that all referenced Flows/Nodes exist in the project - For each Link, validate instances exist and ports are compatible

check_flow(flow_ref) → List[ProjIssue]

Check a specific flow for issues.

Parameters:

flow_ref – Reference of the flow to check.

validate() → ProjectValidationResult

Validate the project data.

Returns:

The result of the project validation, including issues found.

Return type:

ProjectValidationResult

class dal.validation.project_validator.Summary(*, total_issues: int, error_count: int, warning_count: int, scopes_checked: List[str])

Bases: BaseModel

Summary of validation results.

error_count: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[dict[str, FieldInfo]] = {'error_count': FieldInfo(annotation=int, required=True), 'scopes_checked': FieldInfo(annotation=List[str], required=True), 'total_issues': FieldInfo(annotation=int, required=True), 'warning_count': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

scopes_checked: List[str]
total_issues: int
warning_count: int

dal.validation.schema module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Developers: - Moawiya Mograbi (moawiya@mov.ai) - 2022

class dal.validation.schema.Schema(schema_path: Path)

Bases: object

validate(data: dict) → ValidationResult

Validate data against the schema.

Parameters:

data (dict) – The data to be validated.

Returns:

Validation results.

Return type:

ValidationResult

class dal.validation.schema.ValidationResult(*args, **kwargs)

Bases: dict

message: str
status: bool

dal.validation.template module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Developers: - Alexandre Pires (alexandre.pires@mov.ai) - 2020

class dal.validation.template.Template(**kwargs)

Bases: object

Loads a config file from json based on a given schema

Example

protocol_config = {

“name”: str, “parameters”: dict,

}

callback_config = {

“name”: str, “libs”: dict, “file”: str

}

port_config = {

“name”: str, “direction”: str, “protocol”: protocol_config, “callback”: callback_config

}

node_config = {

“logfile”: str, “name”: str, “parameters”: dict, “ports”: [port_config]

}

c = Template.load(“node.json”, node_config, “./tests/configs”)

static load(uri: str, config: dict)

Loads a configuration file, validate against config schema returns a Template object

static load_dict(values: dict, config: dict)

Creates a template from a dictionary

dal.validation.validator module

Copyright (C) Mov.ai - All Rights Reserved Unauthorized copying of this file, via any medium is strictly prohibited Proprietary and confidential

Developers: - Moawiya Mograbi (moawiya@mov.ai) - 2022

class dal.validation.validator.JsonValidator(*args, **kwargs)

Bases: object

Validator responsible to load schema json files and validate files according to it’s type.

VERSION = '2.4'
validate(scope: str, data: dict)

Validate the content against the schema of the given scope.

Parameters:

  • scope (str) – The type of the schema to validate against.

  • data (dict) – The data to validate.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

class dal.validation.validator.POFileValidator

Bases: object

Confirm the PO file has a valid format

static validate(scope: str, data: dict)

Validate the content against the schema of the given scope.

Parameters:

  • scope (str) – The type of the schema to validate against.

  • data (dict) – The data to validate.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

class dal.validation.validator.TranslationValidator

Bases: object

static validate(scope: str, data: dict)

Validate the content against the schema of the given scope.

Parameters:

  • scope (str) – The type of the schema to validate against.

  • data (dict) – The data to validate.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

class dal.validation.validator.Validator(*args, **kwargs)

Bases: Protocol

Protocol for validators that validate data against a schema.

static validate(scope: str, data: dict) → None

Validate the content against the schema of the given scope.

Parameters:

  • scope (str) – The type of the schema to validate against.

  • data (dict) – The data to validate.

Raises:

  • SchemaTypeNotKnown – If the scope is not known to the validator.

  • ValueError – If the data does not conform to the schema.

Module contents