thoth.adviser.python.pipeline package

Subpackages

Submodules

thoth.adviser.python.pipeline.context_base module

A base for a pipeline unit context - stride or step context.

class thoth.adviser.python.pipeline.context_base.ContextBase(stats=None)[source]

Bases: object

A base for a pipeline unit context.

property stats

Retrieve statistics for the current running step.

thoth.adviser.python.pipeline.exceptions module

Exceptions hierarchy used inside stack generation pipeline.

These exceptions should not be propagated outside of pipeline module.

exception thoth.adviser.python.pipeline.exceptions.CannotRemovePackage[source]

Bases: thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase

Raised if the given package cannot be removed from paths due to dependencies.

exception thoth.adviser.python.pipeline.exceptions.NotAICoERelease[source]

Bases: thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase

Raised if the given package is not AICoE release.

exception thoth.adviser.python.pipeline.exceptions.NotResolvedError[source]

Bases: thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase

Raised if trying to run stack generation pipeline without fully resolved dependency.

As produced stacks should be fully resolved, packages which are not resolved yet should be filtered out in the pipeline units.

exception thoth.adviser.python.pipeline.exceptions.PackageNotFound[source]

Bases: thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase

Raised if the given package cannot be found (mostly internal error).

exception thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase[source]

Bases: Exception

An exception raised inside stack generator pipeline.

exception thoth.adviser.python.pipeline.exceptions.StrideRemoveStack[source]

Bases: thoth.adviser.python.pipeline.exceptions.PipelineExceptionBase

Raised from within a stride if the given stack should be removed.

thoth.adviser.python.pipeline.pipeline module

Implementation of stack generation pipeline.

class thoth.adviser.python.pipeline.pipeline.Pipeline(project: thoth.python.project.Project, sieves: List[Tuple[type, dict]] = NOTHING, steps: List[Tuple[type, dict]] = NOTHING, strides: List[Tuple[type, dict]] = NOTHING, graph: thoth.storages.graph.postgres.GraphDatabase = NOTHING, library_usage: dict = NOTHING, solver: thoth.adviser.python.solver.PythonPackageGraphSolver = None, stack_info: List[Dict] = NOTHING)[source]

Bases: object

A stack generation pipeline.

conduct(count: int = None, limit: int = None) → Generator[[thoth.adviser.python.pipeline.product.PipelineProduct, None], None][source]

Chain execution of each step and filter in a pipeline and execute it.

get_configuration() → dict[source]

Get a serialized configuration of this pipeline.

get_stack_info() → List[Dict][source]

Get report of a pipeline run.

This report includes reports from dependency graph walk.

graph_default()[source]

Get default graph instance if no explicitly provided.

property solver

Get solver instance - solver implemented on top of graph database.

thoth.adviser.python.pipeline.product module

Product of stack generation pipeline.

class thoth.adviser.python.pipeline.product.PipelineProduct(project: thoth.python.project.Project, score: float, justification: List[Dict], graph: thoth.storages.graph.postgres.GraphDatabase)[source]

Bases: object

One product of stack generation pipeline.

finalize() → None[source]

Finalize creation of a pipeline product.

thoth.adviser.python.pipeline.sieve module

A base class for implementing sieves to filter out direct dependencies.

class thoth.adviser.python.pipeline.sieve.Sieve(graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: dict = NOTHING, parameters: dict = NOTHING, name: str = None)[source]

Bases: thoth.adviser.python.pipeline.unit_base.PipelineUnitBase

Filter out direct dependencies based on the given criteria.

abstract run(sieve_context: thoth.adviser.python.pipeline.sieve_context.SieveContext) → None[source]

Filter out package versions based on sieve implementation.

thoth.adviser.python.pipeline.sieve_context module

Implementation of sieve context base class.

class thoth.adviser.python.pipeline.sieve_context.SieveContext(stats=None, packages: dict = NOTHING)[source]

Bases: thoth.adviser.python.pipeline.context_base.ContextBase

A wrapper for sieves encapsulating common logic for operations performed by sieves.

classmethod from_package_versions(package_versions: List[thoth.python.package_version.PackageVersion]) → thoth.adviser.python.pipeline.sieve_context.SieveContext[source]

Instantiate from a list of package-versions.

iter_direct_dependencies(develop: bool = None) → Generator[[thoth.python.package_version.PackageVersion, None], None][source]

Iterate over direct dependencies, respect their ordering.

iter_direct_dependencies_tuple() → Generator[[Tuple[str, str, str], None], None][source]

Iterate over direct dependencies, respect their ordering and return a package tuple.

packages
remove_package(package_version: thoth.python.package_version.PackageVersion) → None[source]

Remove package-versions - a syntax sugar for remove_package_tuple.

remove_package_tuple(package_tuple: Tuple[str, str, str]) → None[source]

Remove the given package tuple.

sort_packages(comparision_func: Callable[[thoth.python.package_version.PackageVersion, thoth.python.package_version.PackageVersion], int], reverse: bool = True) → None[source]

Sort packages based on the comparision function.

The sorting is stable, meaning, it reflects previous relative order of packages so its completely fine to perform multiple sorts in sieves - the more recent sieve will have higher priority.

thoth.adviser.python.pipeline.stack_candidates module

Wrapper for transparent manipulation with stack candidates.

class thoth.adviser.python.pipeline.stack_candidates.StackCandidates(input_project: thoth.python.project.Project, direct_dependencies_map: Dict[str, Dict[str, Dict[str, thoth.python.package_version.PackageVersion]]] = NOTHING, transitive_dependencies_map: Dict[str, Dict[str, Dict[str, thoth.python.package_version.PackageVersion]]] = NOTHING, count: int = None, stack_candidates: List[Tuple[float, List[Tuple[str, str, str]]]] = NOTHING)[source]

Bases: object

A wrapper for stack candidates.

add(stride_context: thoth.adviser.python.pipeline.stride_context.StrideContext) → None[source]

Add the given stack candidate to candidate listing.

count
direct_dependencies_map
generate_pipeline_products(graph: thoth.storages.graph.postgres.GraphDatabase) → Generator[[thoth.adviser.python.pipeline.product.PipelineProduct, None], None][source]

Generate projects in stack candidates.

All the candidates are discarded after calling this function.

get_package_version_tuple(package_tuple: tuple) → thoth.python.package_version.PackageVersion[source]

Get package version from the dependencies map based on tuple provided.

input_project
transitive_dependencies_map

thoth.adviser.python.pipeline.stats_base module

Implementation of statistics for a stack generation pipeline and every pipeline step.

class thoth.adviser.python.pipeline.stats_base.StatsBase(start_time: float = NOTHING, unit_start_time: float = None, units_run: dict = NOTHING, current_name: str = None)[source]

Bases: object

Statistics accumulated for a unit or units inside pipeline.

get_duration() → float[source]

Return duration of a pipeline step in seconds measured from start till call of this function.

abstract log_report() → None[source]

Log report for a pipeline unit.

reset_stats(unit: thoth.adviser.python.pipeline.unit_base.PipelineUnitBase) → None[source]

Reset stats respecting the next pipeline unit to be run.

property start_time

Get start time of a pipeline step.

start_timer() → None[source]

Start timer for time based statistics.

thoth.adviser.python.pipeline.step module

Implementation a base class for a step in stack generation pipeline.

class thoth.adviser.python.pipeline.step.Step(graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: dict = NOTHING, parameters: dict = NOTHING, name: str = None)[source]

Bases: thoth.adviser.python.pipeline.unit_base.PipelineUnitBase

A step in a stack generation pipeline.

abstract run(step_context: thoth.adviser.python.pipeline.step_context.StepContext) → None[source]

Entrypoint for a stack pipeline step.

thoth.adviser.python.pipeline.step_context module

An object wrapping stack candidates used to generate stacks from in pipeline.

class thoth.adviser.python.pipeline.step_context.StepContext(packages: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion], unsolved_packages: Dict[Tuple[str, str, None], thoth.adviser.python.pipeline.unsolved_package.UnsolvedPackage], dependency_graph_adaptation: thoth.adviser.python.dependency_graph.adaptation.graph.DependencyGraph = None, stats: thoth.adviser.python.pipeline.step_stats.StepStats = NOTHING)[source]

Bases: thoth.adviser.python.pipeline.context_base.ContextBase

A wrapper representing stack stack candidates.

dependency_graph_adaptation
classmethod from_paths(direct_dependencies: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion], paths: Dict[Tuple[str, str, str], List[Tuple[Tuple[str, str, str], Tuple[str, str, str]]]]) → thoth.adviser.python.pipeline.step_context.StepContext[source]

Instantiate step context from paths.

static is_unsolved_package_tuple(package_tuple: Tuple[str, Optional[str], Optional[str]]) → bool[source]

Check if the given package tuple represents an unsolved package.

iter_all_dependencies(develop: bool = None) → Generator[[thoth.python.package_version.PackageVersion, None], None][source]

Iterate over all possible dependencies, make sure each dependency is returned once.

iter_all_dependencies_tuple() → Generator[[Tuple[str, str, str], None], None][source]

Iterate over all the dependencies, return each as a tuple.

iter_direct_dependencies(develop: bool = None) → Generator[[thoth.python.package_version.PackageVersion, None], None][source]

Iterate over direct dependencies, respect their ordering.

iter_direct_dependencies_tuple() → Generator[[Tuple[str, str, str], None], None][source]

Iterate over direct dependencies, respect their ordering and return a package tuple.

iter_transitive_dependencies(develop: bool = None) → Generator[[Union[thoth.python.package_version.PackageVersion, thoth.adviser.python.pipeline.unsolved_package.UnsolvedPackage], None], None][source]

Iterate over indirect (transitive) dependencies, respect their ordering.

iter_transitive_dependencies_tuple() → Generator[[Tuple[str, str, str], None], None][source]

Iterate over indirect (transitive) dependencies, respect their ordering.

packages
remove_package_tuples(*package_tuples: Tuple[str, str, str]) → thoth.adviser.python.dependency_graph.adaptation.transaction.DependencyGraphTransaction[source]

Remove all packages from all the resolution paths, transactional operation.

score_package_tuple(package_tuple: Tuple[str, str, str], score: float) → None[source]

Score positively or negatively the given package making sure it gets precedence/postponed in resolution.

sort_paths(comparision_func: Callable[[thoth.python.package_version.PackageVersion, thoth.python.package_version.PackageVersion], int], reverse: bool = True) → None[source]

Sort paths in the dependency graph.

property stats

Retrieve statistics kept during steps computation.

unsolved_packages

thoth.adviser.python.pipeline.step_stats module

Implementation of statistics for a stack generation pipeline and every pipeline step.

class thoth.adviser.python.pipeline.step_stats.StepStats(start_time: float = NOTHING, unit_start_time: float = None, units_run: dict = NOTHING, current_name: str = None, packages_removed_count: dict = NOTHING)[source]

Bases: thoth.adviser.python.pipeline.stats_base.StatsBase

Statistics accumulated for steps in pipeline.

log_report() → None[source]

Log report for a pipeline step.

mark_removed_package_tuple(package_tuple: Tuple[str, str, str])[source]

Keep track of packages removed in a step run for statistics.

thoth.adviser.python.pipeline.stride module

Implementation a base class for a stride in stack generation pipeline.

class thoth.adviser.python.pipeline.stride.Stride(graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: dict = NOTHING, parameters: dict = NOTHING, name: str = None)[source]

Bases: thoth.adviser.python.pipeline.unit_base.PipelineUnitBase

A stride in a stack generation pipeline.

abstract run(stride_context: thoth.adviser.python.pipeline.stride_context.StrideContext) → None[source]

Entrypoint for a stride pipeline stride.

thoth.adviser.python.pipeline.stride_context module

Context passed into stride in stack generation pipeline.

class thoth.adviser.python.pipeline.stride_context.StrideContext(stack_candidate: List[Tuple[str, str, str]], score: float = 0.0, justification: list = NOTHING, stats: thoth.adviser.python.pipeline.stride_stats.StrideStats = NOTHING)[source]

Bases: thoth.adviser.python.pipeline.context_base.ContextBase

A context carried during stride scoring and checking.

adjust_score(score: float, justification: List[dict] = None) → None[source]

Adjust score for the given stack candidate, modify justification reported to end user.

property justification

Retrieve a list of justifications.

property score

Get score of stack candidate.

stack_candidate

thoth.adviser.python.pipeline.stride_stats module

Implementation of statistics for generation pipeline.

class thoth.adviser.python.pipeline.stride_stats.StrideStats(start_time: float = NOTHING, unit_start_time: float = None, units_run: dict = NOTHING, current_name: str = None)[source]

Bases: thoth.adviser.python.pipeline.stats_base.StatsBase

Statistics accumulated for strides in pipeline.

log_report() → None[source]

Log report for a pipeline step.

thoth.adviser.python.pipeline.unit_base module

A base class for implementing pipeline units - strides and steps.

class thoth.adviser.python.pipeline.unit_base.PipelineUnitBase(graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: dict = NOTHING, parameters: dict = NOTHING, name: str = None)[source]

Bases: object

A base class for implementing pipeline units - strides and steps.

PARAMETERS_DEFAULT = {}
PARAMETERS_SCHEMA = None
classmethod compute_expanded_parameters(parameters_dict: Optional[dict])[source]

Compute parameters as they would be computed based on unit configuration.

classmethod get_aicoe_configuration(package_version: thoth.python.package_version.PackageVersion) → Optional[dict][source]

Get AICoE specific configuration encoded in the AICoE index URL.

graph
static is_aicoe_release(package_version: thoth.python.package_version.PackageVersion) → bool[source]

Check if the given package-version is AICoE release.

library_usage
property name

Name of a stack pipeline.

property parameters

Get parameters of instantiated pipeline step.

project
abstract run(context: thoth.adviser.python.pipeline.context_base.ContextBase) → None[source]

Entrypoint for a stack pipeline unit.

to_dict() → dict[source]

Turn this pipeline step into its dictionary representation.

update_parameters(parameters_dict: dict) → None[source]

Set parameters for a stack pipeline.

If setting parameters fails due to schema checks, parameters are kept in an invalid state.

thoth.adviser.python.pipeline.unsolved_package module

Representation of an unsolved package in dependency graph.

class thoth.adviser.python.pipeline.unsolved_package.UnsolvedPackage(package_name: str, package_version: str = None, develop: bool = False)[source]

Bases: object

Representation of an unsolved package in the dependency graph.

develop
package_name
package_version
to_tuple() → Tuple[str, str, None][source]

Convert this package to a tuple representation.

To be fully compliant with PackageVersion, provide None index explictly

Module contents

Implementation of stack generation pipeline.