cdm_data_loaders.readers.jsonschema_xsv.xsv_validator.helpers ¶

Helpers for the qsv validator.

ErrorRecord `pydantic-model` ¶

Bases: BaseModel

A single error encountered while cleaning/validating a file.

Used so that errors always contains a uniform type, regardless of whether the error came from a failed qsv subprocess, an OSError raised while copying/moving files, or any other type of fun error that might occur.

Parameters:

Name	Type	Description	Default
`file`	`str`	name of the xsv file being processed when the error occurred	required
`message`	`str`	human-readable description of the error	required
`returncode`	`(int \| None, optional)`	return code of the qsv subprocess, if the error originated from one	required

Show JSON schema:

{
  "description": "A single error encountered while cleaning/validating a file.\n\nUsed so that ``errors`` always contains a uniform type, regardless of whether the\nerror came from a failed qsv subprocess, an ``OSError`` raised while copying/moving files, or\nany other type of fun error that might occur.\n\n:param file: name of the xsv file being processed when the error occurred\n:type file: str\n:param message: human-readable description of the error\n:type message: str\n:param returncode: return code of the qsv subprocess, if the error originated from one\n:type returncode: int | None, optional",
  "properties": {
    "file": {
      "title": "File",
      "type": "string"
    },
    "message": {
      "title": "Message",
      "type": "string"
    },
    "returncode": {
      "anyOf": [
        {
          "type": "integer"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Returncode"
    }
  },
  "required": [
    "file",
    "message"
  ],
  "title": "ErrorRecord",
  "type": "object"
}

Fields:

file (str)
message (str)
returncode (int | None)

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

class ErrorRecord(BaseModel):
    """A single error encountered while cleaning/validating a file.

    Used so that ``errors`` always contains a uniform type, regardless of whether the
    error came from a failed qsv subprocess, an ``OSError`` raised while copying/moving files, or
    any other type of fun error that might occur.

    :param file: name of the xsv file being processed when the error occurred
    :type file: str
    :param message: human-readable description of the error
    :type message: str
    :param returncode: return code of the qsv subprocess, if the error originated from one
    :type returncode: int | None, optional
    """

    file: str
    message: str
    returncode: int | None = None

    @classmethod
    def from_qsv_result(cls, file_name: str, result: subprocess.CompletedProcess[str]) -> Self:
        """Build an `ErrorRecord` from a failed qsv subprocess result."""
        return cls(file=file_name, message=(result.stderr or "").strip(), returncode=result.returncode)

    @classmethod
    def from_exception(cls, file_name: str, exc: Exception) -> Self:
        """Build an `ErrorRecord` from a raised exception (e.g. `OSError`)."""
        return cls(file=file_name, message=str(exc))

    def to_rejected_row(self, table_name: str) -> dict[str, str | int | None]:
        """Format this error record for the ``_rejected`` table."""
        return make_rejected_row(table_name, self.message, source_file=self.file)

from_qsv_result `classmethod` ¶

from_qsv_result(
    file_name: str, result: CompletedProcess[str]
) -> Self

Build an ErrorRecord from a failed qsv subprocess result.

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

@classmethod
def from_qsv_result(cls, file_name: str, result: subprocess.CompletedProcess[str]) -> Self:
    """Build an `ErrorRecord` from a failed qsv subprocess result."""
    return cls(file=file_name, message=(result.stderr or "").strip(), returncode=result.returncode)

from_exception `classmethod` ¶

from_exception(file_name: str, exc: Exception) -> Self

Build an ErrorRecord from a raised exception (e.g. OSError).

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

@classmethod
def from_exception(cls, file_name: str, exc: Exception) -> Self:
    """Build an `ErrorRecord` from a raised exception (e.g. `OSError`)."""
    return cls(file=file_name, message=str(exc))

to_rejected_row ¶

to_rejected_row(
    table_name: str,
) -> dict[str, str | int | None]

Format this error record for the _rejected table.

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def to_rejected_row(self, table_name: str) -> dict[str, str | int | None]:
    """Format this error record for the ``_rejected`` table."""
    return make_rejected_row(table_name, self.message, source_file=self.file)

CleanerValidatorArgs `pydantic-model` ¶

Bases: BaseModel

Class containing required parameters for running the XSV file clean/validate functions.

Parameters:

Name	Type	Description	Default
`qsv_cmd`	`str`	command for executing qsv	required
`xsv_file_path`	`FilePath`	path to the data file to be validated	required
`header_file_path`	`FilePath`	path to the generated header file	required
`first_pass_schema`	`ValidatedXsvSchema`	schema for initial validation of the data	required
`post_norm_schema`	`ValidatedXsvSchema`	schema for validating data after cleaning and normalisation	required
`tmp_dir_path`	`DirectoryPath`	temporary working directory	required
`qsv_output_dir_path`	`DirectoryPath`	directory for qsv output from failed executions	required
`validated_file_dir_path`	`DirectoryPath`	directory to put the cleaned, validated files in	required
`errors`	`list[ErrorRecord]`	list of errors accumulated during processing	required
`delimiter`	`(CharStr, optional)`	delimiter used in the xsv file, defaults to tab	required
`comment_char`	`(CharStr, optional)`	comment character used in the xsv file, defaults to `#`	required
`quote`	`(CharStr, optional)`	quote character used in the xsv file; if unset, qsv will assume it is `"`	required
`escape`	`(NonEmptyStr, optional)`	the escape character used in the xsv file. If not specified, qsv escapes quotes by doubling them. Default None	required
`null_regex`	`(NonEmptyStr, optional)`	regex to replace with an empty string	required
`null_regex_cols`	`list[NonEmptyStr] \| None`	list of columns to apply the null regex to, optional; defaults to None (all cols)	required
`missing_header`	`(bool, optional)`	whether or not the xsv file has a missing header, defaults to False	required
`source_id`	`(NonEmptyStr \| None, optional)`	identifier recorded for the input file in error records, e.g. its path relative to the input location; defaults to the name of `xsv_file_path`	required

Show JSON schema:

{
  "$defs": {
    "ErrorRecord": {
      "description": "A single error encountered while cleaning/validating a file.\n\nUsed so that ``errors`` always contains a uniform type, regardless of whether the\nerror came from a failed qsv subprocess, an ``OSError`` raised while copying/moving files, or\nany other type of fun error that might occur.\n\n:param file: name of the xsv file being processed when the error occurred\n:type file: str\n:param message: human-readable description of the error\n:type message: str\n:param returncode: return code of the qsv subprocess, if the error originated from one\n:type returncode: int | None, optional",
      "properties": {
        "file": {
          "title": "File",
          "type": "string"
        },
        "message": {
          "title": "Message",
          "type": "string"
        },
        "returncode": {
          "anyOf": [
            {
              "type": "integer"
            },
            {
              "type": "null"
            }
          ],
          "default": null,
          "title": "Returncode"
        }
      },
      "required": [
        "file",
        "message"
      ],
      "title": "ErrorRecord",
      "type": "object"
    },
    "ValidatedXsvSchema": {
      "description": "A JSON Schema to be used for parsing and validating XSV data.",
      "properties": {
        "jsonschema": {
          "additionalProperties": true,
          "title": "Jsonschema",
          "type": "object"
        }
      },
      "required": [
        "jsonschema"
      ],
      "title": "ValidatedXsvSchema",
      "type": "object"
    }
  },
  "description": "Class containing required parameters for running the XSV file clean/validate functions.\n\n:param qsv_cmd: command for executing qsv\n:type qsv_cmd: str\n:param xsv_file_path: path to the data file to be validated\n:type xsv_file_path: FilePath\n:param header_file_path: path to the generated header file\n:type header_file_path: FilePath\n\n:param first_pass_schema: schema for initial validation of the data\n:type first_pass_schema: ValidatedXsvSchema\n:param post_norm_schema: schema for validating data after cleaning and normalisation\n:type post_norm_schema: ValidatedXsvSchema\n\n:param tmp_dir_path: temporary working directory\n:type tmp_dir_path: DirectoryPath\n:param qsv_output_dir_path: directory for qsv output from failed executions\n:type qsv_output_dir_path: DirectoryPath\n:param validated_file_dir_path: directory to put the cleaned, validated files in\n:type validated_file_dir_path: DirectoryPath\n:param errors: list of errors accumulated during processing\n:type errors: list[ErrorRecord]\n:param delimiter: delimiter used in the xsv file, defaults to tab\n:type delimiter: CharStr, optional\n:param comment_char: comment character used in the xsv file, defaults to `#`\n:type comment_char: CharStr, optional\n:param quote: quote character used in the xsv file; if unset, qsv will assume it is `\"`\n:type quote: CharStr, optional\n:param escape: the escape character used in the xsv file. If not specified, qsv escapes quotes by doubling them. Default None\n:type escape: NonEmptyStr, optional\n:param null_regex: regex to replace with an empty string\n:type null_regex: NonEmptyStr, optional\n:param null_regex_cols: list of columns to apply the null regex to, optional; defaults to None (all cols)\n:type null_regex_cols: list[NonEmptyStr] | None\n:param missing_header: whether or not the xsv file has a missing header, defaults to False\n:type missing_header: bool, optional\n:param source_id: identifier recorded for the input file in error records, e.g. its path relative to the\n    input location; defaults to the name of ``xsv_file_path``\n:type source_id: NonEmptyStr | None, optional",
  "properties": {
    "qsv_cmd": {
      "minLength": 1,
      "title": "Qsv Cmd",
      "type": "string"
    },
    "xsv_file_path": {
      "format": "file-path",
      "title": "Xsv File Path",
      "type": "string"
    },
    "header_file_path": {
      "format": "file-path",
      "title": "Header File Path",
      "type": "string"
    },
    "first_pass_schema": {
      "$ref": "#/$defs/ValidatedXsvSchema"
    },
    "post_norm_schema": {
      "$ref": "#/$defs/ValidatedXsvSchema"
    },
    "tmp_dir_path": {
      "format": "directory-path",
      "title": "Tmp Dir Path",
      "type": "string"
    },
    "qsv_output_dir_path": {
      "format": "directory-path",
      "title": "Qsv Output Dir Path",
      "type": "string"
    },
    "validated_file_dir_path": {
      "format": "directory-path",
      "title": "Validated File Dir Path",
      "type": "string"
    },
    "errors": {
      "items": {
        "$ref": "#/$defs/ErrorRecord"
      },
      "title": "Errors",
      "type": "array"
    },
    "delimiter": {
      "default": "\t",
      "maxLength": 1,
      "minLength": 1,
      "title": "Delimiter",
      "type": "string"
    },
    "comment_char": {
      "default": "#",
      "maxLength": 1,
      "minLength": 1,
      "title": "Comment Char",
      "type": "string"
    },
    "quote": {
      "anyOf": [
        {
          "maxLength": 1,
          "minLength": 1,
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Quote"
    },
    "escape": {
      "anyOf": [
        {
          "minLength": 1,
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Escape"
    },
    "null_regex": {
      "anyOf": [
        {
          "minLength": 1,
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Null Regex"
    },
    "null_regex_cols": {
      "anyOf": [
        {
          "items": {
            "minLength": 1,
            "type": "string"
          },
          "type": "array"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Null Regex Cols"
    },
    "missing_header": {
      "default": false,
      "title": "Missing Header",
      "type": "boolean"
    },
    "source_id": {
      "anyOf": [
        {
          "minLength": 1,
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Source Id"
    }
  },
  "required": [
    "qsv_cmd",
    "xsv_file_path",
    "header_file_path",
    "first_pass_schema",
    "post_norm_schema",
    "tmp_dir_path",
    "qsv_output_dir_path",
    "validated_file_dir_path"
  ],
  "title": "CleanerValidatorArgs",
  "type": "object"
}

Fields:

qsv_cmd (NonEmptyStr)
xsv_file_path (FilePath)
header_file_path (FilePath)
first_pass_schema (ValidatedXsvSchema)
post_norm_schema (ValidatedXsvSchema)
tmp_dir_path (DirectoryPath)
qsv_output_dir_path (DirectoryPath)
validated_file_dir_path (DirectoryPath)
errors (list[ErrorRecord])
delimiter (CharStr)
comment_char (CharStr)
quote (CharStr | None)
escape (NonEmptyStr | None)
null_regex (NonEmptyStr | None)
null_regex_cols (list[NonEmptyStr] | None)
missing_header (bool)
source_id (NonEmptyStr | None)

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

class CleanerValidatorArgs(BaseModel):
    """Class containing required parameters for running the XSV file clean/validate functions.

    :param qsv_cmd: command for executing qsv
    :type qsv_cmd: str
    :param xsv_file_path: path to the data file to be validated
    :type xsv_file_path: FilePath
    :param header_file_path: path to the generated header file
    :type header_file_path: FilePath

    :param first_pass_schema: schema for initial validation of the data
    :type first_pass_schema: ValidatedXsvSchema
    :param post_norm_schema: schema for validating data after cleaning and normalisation
    :type post_norm_schema: ValidatedXsvSchema

    :param tmp_dir_path: temporary working directory
    :type tmp_dir_path: DirectoryPath
    :param qsv_output_dir_path: directory for qsv output from failed executions
    :type qsv_output_dir_path: DirectoryPath
    :param validated_file_dir_path: directory to put the cleaned, validated files in
    :type validated_file_dir_path: DirectoryPath
    :param errors: list of errors accumulated during processing
    :type errors: list[ErrorRecord]
    :param delimiter: delimiter used in the xsv file, defaults to tab
    :type delimiter: CharStr, optional
    :param comment_char: comment character used in the xsv file, defaults to `#`
    :type comment_char: CharStr, optional
    :param quote: quote character used in the xsv file; if unset, qsv will assume it is `"`
    :type quote: CharStr, optional
    :param escape: the escape character used in the xsv file. If not specified, qsv escapes quotes by doubling them. Default None
    :type escape: NonEmptyStr, optional
    :param null_regex: regex to replace with an empty string
    :type null_regex: NonEmptyStr, optional
    :param null_regex_cols: list of columns to apply the null regex to, optional; defaults to None (all cols)
    :type null_regex_cols: list[NonEmptyStr] | None
    :param missing_header: whether or not the xsv file has a missing header, defaults to False
    :type missing_header: bool, optional
    :param source_id: identifier recorded for the input file in error records, e.g. its path relative to the
        input location; defaults to the name of ``xsv_file_path``
    :type source_id: NonEmptyStr | None, optional
    """

    qsv_cmd: NonEmptyStr

    xsv_file_path: FilePath
    header_file_path: FilePath

    first_pass_schema: ValidatedXsvSchema
    post_norm_schema: ValidatedXsvSchema

    tmp_dir_path: DirectoryPath
    qsv_output_dir_path: DirectoryPath
    validated_file_dir_path: DirectoryPath

    errors: list[ErrorRecord] = Field(default_factory=lambda _: [])

    delimiter: CharStr = "\t"
    comment_char: CharStr = "#"
    quote: CharStr | None = None
    escape: NonEmptyStr | None = None
    null_regex: NonEmptyStr | None = None
    null_regex_cols: list[NonEmptyStr] | None = None
    missing_header: bool = False
    source_id: NonEmptyStr | None = None

    @computed_field
    @cached_property
    def qsv_env(self: Self) -> dict[str, str]:
        """Environment for qsv commands to run in."""
        return {**os.environ, "QSV_COMMENT_CHAR": self.comment_char}

    @computed_field
    @property
    def xsv_file_base_name(self: Self) -> str:
        """Base name of the input xsv file, without the final suffix."""
        return self.xsv_file_path.stem

    @computed_field
    @property
    def file_name(self) -> str:
        """Identifier of the input xsv file used in error records: ``source_id`` if set, else the file name."""
        return self.source_id or self.xsv_file_path.name

    @computed_field
    @property
    def ext(self) -> str:
        """Extension for the xsv files; includes the `.`."""
        return SEP_TO_EXT.get(self.delimiter) or self.xsv_file_path.suffix

    @computed_field
    @property
    def valid_file_suffix(self) -> str:
        """Suffix added to validator output file containing valid lines."""
        return f"valid{self.ext}"

    @computed_field
    @property
    def invalid_file_suffix(self) -> str:
        """Suffix added to validator output file containing valid lines."""
        return f"invalid{self.ext}"

qsv_env `cached` `property` ¶

qsv_env: dict[str, str]

Environment for qsv commands to run in.

xsv_file_base_name `property` ¶

xsv_file_base_name: str

Base name of the input xsv file, without the final suffix.

file_name `property` ¶

file_name: str

Identifier of the input xsv file used in error records: source_id if set, else the file name.

ext `property` ¶

ext: str

Extension for the xsv files; includes the ..

valid_file_suffix `property` ¶

valid_file_suffix: str

Suffix added to validator output file containing valid lines.

invalid_file_suffix `property` ¶

invalid_file_suffix: str

Suffix added to validator output file containing valid lines.

FileNames `dataclass` ¶

Dataclass for storing derived file names.

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

@dataclass
class FileNames:
    """Dataclass for storing derived file names."""

    valid_output: Path
    errors: Path
    valid_lines: Path
    invalid_lines: Path
    schema: Path

copy_safely ¶

copy_safely(
    src: Path, dst: Path, args: CleanerValidatorArgs
) -> bool

Copy src to dst, recording an ErrorRecord and returning False on failure.

Parameters:

Name	Type	Description	Default
`src`	`Path`	source path	required
`dst`	`Path`	destination path	required
`args`	`CleanerValidatorArgs`	args for the cleaner validator	required

Returns:

Type	Description
`bool`	True if the copy succeeded, False otherwise

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def copy_safely(src: Path, dst: Path, args: "CleanerValidatorArgs") -> bool:
    """Copy `src` to `dst`, recording an `ErrorRecord` and returning False on failure.

    :param src: source path
    :type src: Path
    :param dst: destination path
    :type dst: Path
    :param args: args for the cleaner validator
    :type args: CleanerValidatorArgs
    :return: True if the copy succeeded, False otherwise
    :rtype: bool
    """
    return _transfer_safely(copy, src, dst, args)

move_safely ¶

move_safely(
    src: Path, dst: Path, args: CleanerValidatorArgs
) -> bool

Move src to dst, recording an ErrorRecord and returning False on failure.

Parameters:

Name	Type	Description	Default
`src`	`Path`	source path	required
`dst`	`Path`	destination path	required
`args`	`CleanerValidatorArgs`	args for the cleaner validator	required

Returns:

Type	Description
`bool`	True if the move succeeded, False otherwise

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def move_safely(src: Path, dst: Path, args: "CleanerValidatorArgs") -> bool:
    """Move `src` to `dst`, recording an `ErrorRecord` and returning False on failure.

    :param src: source path
    :type src: Path
    :param dst: destination path
    :type dst: Path
    :param args: args for the cleaner validator
    :type args: CleanerValidatorArgs
    :return: True if the move succeeded, False otherwise
    :rtype: bool
    """
    return _transfer_safely(move, src, dst, args)

non_header_lines_present ¶

non_header_lines_present(path: Path) -> bool

Return True if path has any non-blank lines beyond the header line.

Stops reading as soon as a second non-blank line is found.

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def non_header_lines_present(path: Path) -> bool:
    """Return True if `path` has any non-blank lines beyond the header line.

    Stops reading as soon as a second non-blank line is found.
    """
    non_blank_lines_seen = 0
    with path.open() as f:
        for line in f:
            if line.strip():
                non_blank_lines_seen += 1
                if non_blank_lines_seen > 1:
                    return True
    return False

prepend_header ¶

prepend_header(
    header_file_path: Path, xsv_file_path: Path, dest: Path
) -> None

Write a new file at dest consisting of the header file's contents followed by the data file's contents.

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def prepend_header(header_file_path: Path, xsv_file_path: Path, dest: Path) -> None:
    """Write a new file at `dest` consisting of the header file's contents followed by the data file's contents."""
    with dest.open("wb") as out:
        out.write(header_file_path.read_bytes())
        out.write(xsv_file_path.read_bytes())

generate_qsv_validate_file_names ¶

generate_qsv_validate_file_names(
    args: CleanerValidatorArgs,
    input_file_name: str,
    first_pass: bool = True,
) -> FileNames

Compute the file names run_qsv_validate generates from a file with validation errors.

Parameters:

Name	Type	Description	Default
`args`	`CleanerValidatorArgs`	arguments	required
`input_file_name`	`str`	input file name	required
`first_pass`	`bool`	whether or not this is the first validation pass, defaults to True	`True`

Returns:

Type	Description
`FileNames`	dataclass containing derived paths (valid_output, errors, valid_lines, invalid_lines, schema)

Source code in src/cdm_data_loaders/readers/jsonschema_xsv/xsv_validator/helpers.py

def generate_qsv_validate_file_names(
    args: CleanerValidatorArgs,
    input_file_name: str,
    first_pass: bool = True,  # noqa: FBT001, FBT002
) -> FileNames:
    """Compute the file names run_qsv_validate generates from a file with validation errors.

    :param args: arguments
    :type args: CleanerValidatorArgs
    :param input_file_name: input file name
    :type input_file_name: str
    :param first_pass: whether or not this is the first validation pass, defaults to True
    :type first_pass: bool, optional
    :return: dataclass containing derived paths (valid_output, errors, valid_lines, invalid_lines, schema)
    :rtype: FileNames
    """
    # file name stem for validated lines and schema: {file_base_name}-header-valid.ext
    output_file_stem = (
        f"{args.xsv_file_base_name}-first-pass" if first_pass else f"{args.xsv_file_base_name}-normalised-validated"
    )

    return FileNames(
        valid_output=args.tmp_dir_path / f"{output_file_stem}{VALID_SUFFIX}{args.ext}",
        errors=args.tmp_dir_path / f"{input_file_name}{VALIDATION_ERRORS}",
        valid_lines=args.tmp_dir_path / f"{input_file_name}.{args.valid_file_suffix}",
        invalid_lines=args.tmp_dir_path / f"{input_file_name}.{args.invalid_file_suffix}",
        schema=args.tmp_dir_path / f"{output_file_stem}.jsonschema.json",
    )

cdm_data_loaders.readers.jsonschema_xsv.xsv_validator.helpers ¶

ErrorRecord pydantic-model ¶

from_qsv_result classmethod ¶

from_exception classmethod ¶

to_rejected_row ¶

CleanerValidatorArgs pydantic-model ¶

qsv_env cached property ¶

xsv_file_base_name property ¶

file_name property ¶

ext property ¶

valid_file_suffix property ¶

invalid_file_suffix property ¶

FileNames dataclass ¶

copy_safely ¶

move_safely ¶

non_header_lines_present ¶

prepend_header ¶

generate_qsv_validate_file_names ¶

ErrorRecord `pydantic-model` ¶

from_qsv_result `classmethod` ¶

from_exception `classmethod` ¶

CleanerValidatorArgs `pydantic-model` ¶

qsv_env `cached` `property` ¶

xsv_file_base_name `property` ¶

file_name `property` ¶

ext `property` ¶

valid_file_suffix `property` ¶

invalid_file_suffix `property` ¶

FileNames `dataclass` ¶