thoth.adviser.prescription.v1 package¶

Submodules¶

thoth.adviser.prescription.v1.add_package_step module¶

A prescription step implementing adding a package to a dependency graph.

class thoth.adviser.prescription.v1.add_package_step.AddPackageStepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.step.StepPrescription

Add package prescription step unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': Any(<Schema({'multi_package_resolution': <class 'bool'>, 'package_version': <Schema({'name': <function _python_package_name>, 'locked_version': <function _locked_version>, 'index_url': All(<class 'str'>, Length(min=1, max=None), msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → None[source]¶

Run main entry-point for steps to skip packages.

thoth.adviser.prescription.v1.boot module¶

A prescription for boot units.

class thoth.adviser.prescription.v1.boot.BootPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Boot prescription implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': Any(<Schema({'package_name': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'run': <Schema({'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_boot_unit_type() → bool[source]¶

Check if this unit is of type boot.

run() → None[source]¶

Run main entry-point for boot units.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.gh_release_notes module¶

GitHub release notes pipeline unit.

class thoth.adviser.prescription.v1.gh_release_notes.GHReleaseNotesWrapPrescription(*, unit_run: bool = False, stack_info_run: bool = False, prescription: Dict[str, Any] = NOTHING, configuration: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

GitHub release notes pipeline unit.

CONFIGURATION_DEFAULT = {'package_name': None, 'package_version': None, 'release_notes': None}¶

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'release_notes': <Schema({'organization': All(<class 'str'>, Length(min=1, max=None), msg=None), 'tag_version_prefix': All(<class 'str'>, Length(min=1, max=None), msg=None), 'repository': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, 'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_wrap_unit_type() → bool[source]¶

Check if this unit is of type wrap.

run(state: thoth.adviser.state.State) → None[source]¶

Add release information to justification for selected packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Include this pipeline unit.

thoth.adviser.prescription.v1.prescription module¶

Implementation of a prescription abstraction.

class thoth.adviser.prescription.v1.prescription.Prescription(*, prescriptions: List[Tuple[str, str]] = NOTHING, boots_dict: Dict[str, Dict[str, Any]] = NOTHING, pseudonyms_dict: Dict[str, Dict[str, Any]] = NOTHING, sieves_dict: Dict[str, Dict[str, Any]] = NOTHING, steps_dict: Dict[str, Dict[str, Any]] = NOTHING, strides_dict: Dict[str, Dict[str, Any]] = NOTHING, wraps_dict: Dict[str, Dict[str, Any]] = NOTHING)[source]¶

Bases: object

Dynamically create pipeline units based on inscription.

boots_dict¶

classmethod from_dict(prescription: Dict[str, Any], *, prescription_instance: Optional[Prescription] = None, prescription_name: str, prescription_release: str) → thoth.adviser.prescription.v1.prescription.Prescription[source]¶

Instantiate prescription from a dictionary representation.

If an instance is provided, a safe merge will be performed.

is_empty() → bool[source]¶

Check if no prescription units are loaded.

iter_boot_units() → Generator[Type[BootType], None, None][source]¶

Iterate over prescription boot units registered in the prescription supplied.

iter_pseudonym_units() → Generator[Type[PseudonymType], None, None][source]¶

Iterate over prescription pseudonym units registered in the prescription supplied.

iter_sieve_units() → Generator[Type[SieveType], None, None][source]¶

Iterate over prescription sieve units registered in the prescription supplied.

iter_step_units() → Generator[Type[StepType], None, None][source]¶

Iterate over prescription step units registered in the prescription supplied.

iter_stride_units() → Generator[Type[StrideType], None, None][source]¶

Iterate over prescription stride units registered in the prescription supplied.

iter_wrap_units() → Generator[Type[WrapType], None, None][source]¶

Iterate over prescription stride units registered in the prescription supplied.

classmethod load(*prescriptions: str) → thoth.adviser.prescription.v1.prescription.Prescription[source]¶

Load prescription from files or from their YAML representation.

prescriptions¶

pseudonyms_dict¶

sieves_dict¶

steps_dict¶

strides_dict¶

property units¶

Iterate over units.

classmethod validate(prescriptions: str) → thoth.adviser.prescription.v1.prescription.Prescription[source]¶

Validate the given prescription.

wraps_dict¶

thoth.adviser.prescription.v1.pseudonym module¶

A base class for implementing package pseudonyms.

class thoth.adviser.prescription.v1.pseudonym.PseudonymPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Pseudonym base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'yield': <Schema({'yield_matched_version': <class 'bool'>, 'package_version': <Schema({'name': <function _python_package_name>, 'locked_version': <function _locked_version>, 'index_url': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_pseudonym_unit_type() → bool[source]¶

Check if this unit is of type pseudonym.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(package_version: thoth.python.package_version.PackageVersion) → Generator[Tuple[str, str, str], None, None][source]¶

Run main entry-point for pseudonyms to map packages to their counterparts.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.schema module¶

JSON schema for pipeline unit prescription in version v1.

thoth.adviser.prescription.v1.sieve module¶

A base class for implementing sieves.

class thoth.adviser.prescription.v1.sieve.SievePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Sieve base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': Any(<Schema({'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_sieve_unit_type() → bool[source]¶

Check if this unit is of type sieve.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(package_versions: Generator[thoth.python.package_version.PackageVersion, None, None]) → Generator[thoth.python.package_version.PackageVersion, None, None][source]¶

Run main entry-point for sieves to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.skip_package_sieve module¶

A prescription sieve implementing skipping a package in a dependency graph.

class thoth.adviser.prescription.v1.skip_package_sieve.SkipPackageSievePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Skip package sieve prescription unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'run': Any(<Schema({'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'match': <Schema({'package_name': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_sieve_unit_type() → bool[source]¶

Check if this unit is of type sieve.

pre_run() → None[source]¶

Initialize this unit before each run.

run(_: Generator[thoth.python.package_version.PackageVersion, None, None]) → Generator[thoth.python.package_version.PackageVersion, None, None][source]¶

Run main entry-point for sieves to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.skip_package_step module¶

A prescription step implementing skipping a package in a dependency graph.

class thoth.adviser.prescription.v1.skip_package_step.SkipPackageStepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.step.StepPrescription

Skip package prescription step unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': Any(<Schema({'multi_package_resolution': <class 'bool'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → None[source]¶

Run main entry-point for steps to skip packages.

thoth.adviser.prescription.v1.step module¶

A base class for implementing steps.

class thoth.adviser.prescription.v1.step.StepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Step base class implementation.

Configuration option multi_package_resolution states whether a step should be run if package is resolved multiple times for the same stack.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': <Schema({'score': <class 'float'>, 'justification': All([<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>, 'package_name': <function _python_package_name>, 'advisory': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_id': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_name': All(<class 'str'>, Length(min=1, max=None), msg=None), 'version_range': <function _specifier_set>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'multi_package_resolution': <class 'bool'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

SCORE_MAX = 1.0¶

SCORE_MIN = -1.0¶

static is_step_unit_type() → bool[source]¶

Check if this unit is of type step.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → Optional[Tuple[Optional[float], Optional[List[Dict[str, str]]]]][source]¶

Run main entry-point for steps to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.stride module¶

A base class for implementing strides.

class thoth.adviser.prescription.v1.stride.StridePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Stride base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'state': <Schema({'resolved_dependencies': [<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>]}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_stride_unit_type() → bool[source]¶

Check if this unit is of type stride.

run(state: thoth.adviser.state.State) → None[source]¶

Run main entry-point for strides.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

thoth.adviser.prescription.v1.unit module¶

A base class for prescription based pipeline units.

class thoth.adviser.prescription.v1.unit.UnitPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.unit.Unit

A base class for implementing pipeline units based on prescription supplied.

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'match': <class 'object'>, 'run': <class 'object'>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

SHOULD_INCLUDE_CACHE = {}¶

classmethod get_unit_name() → str[source]¶

Get the name of the current prescription unit.

This method is a class method and MUST NOT be used when obtaining unit name on an instance. As part of the memory optimization we use class to get the current name of a prescription unit with assigned prescription. This means that the prescription unit instance would have different names reported with this method based on the current class context.

property match_prescription¶

Get match part of the prescription assigned.

property name¶

Get name of the prescription instance.

pre_run() → None[source]¶

Prepare this pipeline unit before running it.

prescription¶

property run_prescription¶

Get run part of the prescription assigned.

classmethod set_prescription(prescription: Dict[str, Any]) → None[source]¶

Set prescription to the unit.

thoth.adviser.prescription.v1.unit_cache module¶

Handle caching of pipeline units.

thoth.adviser.prescription.v1.unit_cache.should_include_cache(func: SHOULD_INCLUDE_FUNC_TYPE) → SHOULD_INCLUDE_FUNC_TYPE[source]¶

Handle caching for parts of prescription units that do not change during pipeline construction.

thoth.adviser.prescription.v1.wrap module¶

A base class for implementing wrap units.

class thoth.adviser.prescription.v1.wrap.WrapPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Wrap base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'state': <Schema({'resolved_dependencies': [<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>]}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'justification': All([<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>, 'package_name': <function _python_package_name>, 'advisory': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_id': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_name': All(<class 'str'>, Length(min=1, max=None), msg=None), 'version_range': <function _specifier_set>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'advised_manifest_changes': <class 'object'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_wrap_unit_type() → bool[source]¶

Check if this unit is of type wrap.

pre_run() → None[source]¶

Prepare this pipeline unit before run.

run(state: thoth.adviser.state.State) → None[source]¶

Run main entry-point for wrap units to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

Module contents¶

Schema v1 based prescription units.

class thoth.adviser.prescription.v1.AddPackageStepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.step.StepPrescription

Add package prescription step unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': Any(<Schema({'multi_package_resolution': <class 'bool'>, 'package_version': <Schema({'name': <function _python_package_name>, 'locked_version': <function _locked_version>, 'index_url': All(<class 'str'>, Length(min=1, max=None), msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → None[source]¶

Run main entry-point for steps to skip packages.

class thoth.adviser.prescription.v1.BootPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Boot prescription implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': Any(<Schema({'package_name': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'run': <Schema({'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_boot_unit_type() → bool[source]¶

Check if this unit is of type boot.

run() → None[source]¶

Run main entry-point for boot units.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.GHReleaseNotesWrapPrescription(*, unit_run: bool = False, stack_info_run: bool = False, prescription: Dict[str, Any] = NOTHING, configuration: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

GitHub release notes pipeline unit.

CONFIGURATION_DEFAULT = {'package_name': None, 'package_version': None, 'release_notes': None}¶

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'release_notes': <Schema({'organization': All(<class 'str'>, Length(min=1, max=None), msg=None), 'tag_version_prefix': All(<class 'str'>, Length(min=1, max=None), msg=None), 'repository': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, 'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_wrap_unit_type() → bool[source]¶

Check if this unit is of type wrap.

run(state: thoth.adviser.state.State) → None[source]¶

Add release information to justification for selected packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Include this pipeline unit.

class thoth.adviser.prescription.v1.PseudonymPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Pseudonym base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'yield': <Schema({'yield_matched_version': <class 'bool'>, 'package_version': <Schema({'name': <function _python_package_name>, 'locked_version': <function _locked_version>, 'index_url': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_pseudonym_unit_type() → bool[source]¶

Check if this unit is of type pseudonym.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(package_version: thoth.python.package_version.PackageVersion) → Generator[Tuple[str, str, str], None, None][source]¶

Run main entry-point for pseudonyms to map packages to their counterparts.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.SievePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Sieve base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': Any(<Schema({'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_sieve_unit_type() → bool[source]¶

Check if this unit is of type sieve.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(package_versions: Generator[thoth.python.package_version.PackageVersion, None, None]) → Generator[thoth.python.package_version.PackageVersion, None, None][source]¶

Run main entry-point for sieves to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.SkipPackageSievePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Skip package sieve prescription unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'run': Any(<Schema({'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'match': <Schema({'package_name': All(<class 'str'>, Length(min=1, max=None), msg=None)}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_sieve_unit_type() → bool[source]¶

Check if this unit is of type sieve.

pre_run() → None[source]¶

Initialize this unit before each run.

run(_: Generator[thoth.python.package_version.PackageVersion, None, None]) → Generator[thoth.python.package_version.PackageVersion, None, None][source]¶

Run main entry-point for sieves to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.SkipPackageStepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.step.StepPrescription

Skip package prescription step unit implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': Any(<Schema({'multi_package_resolution': <class 'bool'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, None, msg=None), 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → None[source]¶

Run main entry-point for steps to skip packages.

class thoth.adviser.prescription.v1.StepPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Step base class implementation.

Configuration option multi_package_resolution states whether a step should be run if package is resolved multiple times for the same stack.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'package_version': <Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>, 'state': <Schema({'resolved_dependencies': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from': All([<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'package_version_from_allow_other': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'multi_package_resolution': <class 'bool'>, 'run': <Schema({'score': <class 'float'>, 'justification': All([<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>, 'package_name': <function _python_package_name>, 'advisory': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_id': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_name': All(<class 'str'>, Length(min=1, max=None), msg=None), 'version_range': <function _specifier_set>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'multi_package_resolution': <class 'bool'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

SCORE_MAX = 1.0¶

SCORE_MIN = -1.0¶

static is_step_unit_type() → bool[source]¶

Check if this unit is of type step.

pre_run() → None[source]¶

Prepare before running this pipeline unit.

run(state: thoth.adviser.state.State, package_version: thoth.python.package_version.PackageVersion) → Optional[Tuple[Optional[float], Optional[List[Dict[str, str]]]]][source]¶

Run main entry-point for steps to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.StridePrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Stride base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'state': <Schema({'resolved_dependencies': [<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>]}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'not_acceptable': All(<class 'str'>, Length(min=1, max=None), msg=None), 'eager_stop_pipeline': All(<class 'str'>, Length(min=1, max=None), msg=None), 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_stride_unit_type() → bool[source]¶

Check if this unit is of type stride.

run(state: thoth.adviser.state.State) → None[source]¶

Run main entry-point for strides.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.

class thoth.adviser.prescription.v1.UnitPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.unit.Unit

A base class for implementing pipeline units based on prescription supplied.

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>, 'match': <class 'object'>, 'run': <class 'object'>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

SHOULD_INCLUDE_CACHE = {}¶

classmethod get_unit_name() → str[source]¶

Get the name of the current prescription unit.

This method is a class method and MUST NOT be used when obtaining unit name on an instance. As part of the memory optimization we use class to get the current name of a prescription unit with assigned prescription. This means that the prescription unit instance would have different names reported with this method based on the current class context.

property match_prescription¶

Get match part of the prescription assigned.

property name¶

Get name of the prescription instance.

pre_run() → None[source]¶

Prepare this pipeline unit before running it.

prescription¶

property run_prescription¶

Get run part of the prescription assigned.

classmethod set_prescription(prescription: Dict[str, Any]) → None[source]¶

Set prescription to the unit.

class thoth.adviser.prescription.v1.WrapPrescription(*, unit_run: bool = False, stack_info_run: bool = False, configuration: Dict[str, Any] = NOTHING, prescription: Dict[str, Any] = NOTHING)[source]¶

Bases: thoth.adviser.prescription.v1.unit.UnitPrescription

Wrap base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'match': <Schema({'state': <Schema({'resolved_dependencies': [<Schema({'name': <function _python_package_name>, 'version': <function _specifier_set>, 'index_url': Any(All(<class 'str'>, Length(min=1, max=None), msg=None), {'not': All(<class 'str'>, Length(min=1, max=None), msg=None)}, msg=None), 'develop': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>]}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'run': <Schema({'justification': All([<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>, 'package_name': <function _python_package_name>, 'advisory': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_id': All(<class 'str'>, Length(min=1, max=None), msg=None), 'cve_name': All(<class 'str'>, Length(min=1, max=None), msg=None), 'version_range': <function _specifier_set>}, extra=PREVENT_EXTRA, required=False) object>], Length(min=1, max=None), msg=None), 'advised_manifest_changes': <class 'object'>, 'stack_info': [<Schema({'type': Any('WARNING', 'INFO', 'ERROR', msg=None), 'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'link': <function _justification_link>}, extra=PREVENT_EXTRA, required=False) object>], 'log': <Schema({'message': All(<class 'str'>, Length(min=1, max=None), msg=None), 'type': Any('WARNING', 'INFO', 'ERROR', msg=None)}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>, 'prescription': <Schema({'run': <class 'bool'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>¶

static is_wrap_unit_type() → bool[source]¶

Check if this unit is of type wrap.

pre_run() → None[source]¶

Prepare this pipeline unit before run.

run(state: thoth.adviser.state.State) → None[source]¶

Run main entry-point for wrap units to filter and score packages.

classmethod should_include(builder_context: PipelineBuilderContext) → Generator[Dict[str, Any], None, None][source]¶

Check if the given pipeline unit should be included in the given pipeline configuration.