Checks

pelican.util.checks.get_empty_result_resource(version=1.0)[source]

Initialize a compiled release-level check result.

Parameters:

version (float) – the check’s version

Return type:

dict[str, Any]

pelican.util.checks.get_empty_result_dataset(version=1.0)[source]

Initialize a dataset-level check result.

Parameters:

version (float) – the check’s version

Return type:

dict[str, Any]

pelican.util.checks.get_empty_result_time_based(version=1.0)[source]

Initialize a time-based check result.

Parameters:

version (float) – the check’s version

Return type:

dict[str, Any]

pelican.util.checks.get_empty_result_time_based_scope()[source]

Initialize a time-based check result accumulator.

Return type:

dict[str, Any]

pelican.util.checks.complete_result_resource(result, application_count, pass_count, reason=None, failed_paths=None)[source]

Build a compiled release-level check result.

Parameters:
  • result (dict[str, Any]) – the check result

  • application_count (int) – the number of times the check was applied

  • pass_count (int) – the number of times the check passed

  • reason (str | None) – the reason to provide if the check was not applied

  • failed_paths (Sequence[str | dict[str, Any]] | None) – the failed paths if the check failed

Return type:

dict[str, Any]

pelican.util.checks.complete_result_resource_pass_fail(result, passed, meta=None)[source]

Build a compiled release-level check result, for a pass-fail check.

Parameters:
  • result (dict[str, Any]) – the check result

  • passed (bool) – whether the check passed

  • meta (dict[str, Any] | None) – the additional data to provide if the check failed

Return type:

dict[str, Any]

pelican.util.checks.field_coverage_check(name, test, version=1.0)[source]
Parameters:
  • name (str) – the machine name of the check

  • test (Callable[[dict[str, Any], str], tuple[bool, str]]) – a function that accepts a dict and a key and returns a tuple of a boolean (whether the test passed) and a string (the reason for any failed test)

  • version (float) – the version number of the check

Return type:

Callable[[dict[str, Any], str], dict[str, Any]]

pelican.util.checks.field_quality_check(name, test, version=1.0, require_type=None, return_value=None)[source]
Parameters:
  • name (str) – the machine name of the check

  • test (Callable[[Any], tuple[bool, str]]) – a function that accepts a value and returns a tuple of a boolean (whether the test passed) and a string (the reason for any failed test)

  • version (float) – the version number of the check

  • require_type (type[Any] | None) – the type that the value must have for the test to run without error

  • return_value (Callable[[Any], Any] | None) – a function that accepts a value and returns the value to set in the returned object

Return type:

Callable[[dict[str, Any], str], dict[str, Any]]

pelican.util.checks.coherent_dates_check(version, pairs)[source]

Return a compiled release-level check result for coherent date pairs.

A pair of dates is coherent if the first date is less than or equal to the second date.

Parameters:
  • version (float) – the check’s version

  • pairs (list[tuple[dict[str, Any], dict[str, Any]]]) – date value pairs

Return type:

dict[str, Any]

class pelican.util.checks.ReservoirSampler(limit)[source]
Parameters:

limit (int)

process(value)[source]
Parameters:

value (Any)

Return type:

None