thoth.adviser package

Subpackages

• thoth.adviser.boots package
  • Submodules
  • thoth.adviser.boots.fully_specified_environment module
  • thoth.adviser.boots.solved_software_environment module
  • thoth.adviser.boots.ubi module
  • thoth.adviser.boots.version_clash module
  • Module contents
• thoth.adviser.predictors package
  • Submodules
  • thoth.adviser.predictors.annealing module
  • thoth.adviser.predictors.hill_climbing module
  • thoth.adviser.predictors.latest module
  • thoth.adviser.predictors.random_walk module
  • thoth.adviser.predictors.sampling module
  • Module contents
• thoth.adviser.sieves package
  • Submodules
  • thoth.adviser.sieves.index_enabled module
  • thoth.adviser.sieves.locked module
  • thoth.adviser.sieves.prereleases module
  • thoth.adviser.sieves.solved module
  • thoth.adviser.sieves.version_constraint module
  • Module contents
• thoth.adviser.steps package
  • Submodules
  • thoth.adviser.steps.cve module
  • thoth.adviser.steps.dropout module
  • Module contents
• thoth.adviser.strides package
  • Submodules
  • thoth.adviser.strides.one_version module
  • thoth.adviser.strides.random_decision module
  • Module contents
• thoth.adviser.wraps package
  • Module contents

Submodules

thoth.adviser.beam module

Implementation of Beam for beam searching part of adviser.

class thoth.adviser.beam.Beam(width: Optional[int] = None, states: List[Tuple[Tuple[float, int], thoth.adviser.state.State]] = NOTHING, states_idx: Dict[int, int] = NOTHING, last_added: Optional[thoth.adviser.state.State] = None, top: Optional[thoth.adviser.state.State] = None, *, keep_history=None, beam_history: List[Tuple[int, Optional[float]]] = NOTHING)

Bases: object

Beam implementation.

The implementation of beam respecting beam width: https://en.wikipedia.org/wiki/Beam_search

Beam is internally implemented on top of a min-heap queue to optimize inserts and respect beam width in O(log(N)).

The most frequent operation performed on the beam is not always the same - it depends on the beam width and number of states generated. If number of states generated is more than the width of the beam, its reasonable to optimize insertions into the beam with checks on beam width.

If number of states generated is smaller than the beam width, the beam could be optimized for removal of states.

As the first scenario is seen in a real-world deployment, the beam uses min-heapq to keep addition to the beam with beam_width checks in O(log(N)) and removals of the states in O(log(N)). To satisfy removals in O(log(N)), the beam maintains a dictionary mapping a state to its index in the beam.

add_state(key: object, state: thoth.adviser.state.State) → None

Add state to the internal state listing (do it in O(log(N)) time.

get(idx: int) → thoth.adviser.state.State

Get i-th element from the beam (constant time), keep it in the beam.

This method is suitable for random state indexing in the beam (like in case of adaptive simulated annealing). The actual index is not into a sorted array and has no special meaning assigned - beam under the hood uses min-heapq (as of now), but the index used is not guaranteed to point to a heap-like data structure.

get_by_id(state_id: int) → thoth.adviser.state.State

Get a state by its id.

get_last() → Optional[thoth.adviser.state.State]

Get state that was added in the previous resolution round.

get_random() → thoth.adviser.state.State

Get a random state from beam.

iter_states() → Generator[[thoth.adviser.state.State, None], None]

Iterate over states, do not respect their score in order of iteration.

iter_states_sorted(reverse: bool = True) → Generator[[thoth.adviser.state.State, None], None]

Iterate over sorted states.

keep_history

new_iteration() → None

Called once a new iteration is done in resolver.

Used to keep track of beam history and to keep track of states added.

plot() → matplotlib.figure.Figure

Plot temperature history of adaptive simulated annealing.

pop(idx: Optional[int] = None) → thoth.adviser.state.State

Pop i-th element from the beam and remove it from the beam (this is actually toppop).

If index is not provided, pop one of the largest elements kept in the beam.

If top_state is pre-cached or idx is explicitly set, this operation is done in O(log(N)), O(N) otherwise.

remove(state: thoth.adviser.state.State) → None

Remove the given state from beam.

reset() → None

Recompute internal state of the beam if any changes to the score of states were made.

property size

Get the current size of beam.

top() → thoth.adviser.state.State

Return the highest rated state as kept in the beam.

This operation is done in O(N) if top is not pre-cached, O(1) otherwise.

width

wipe() → None

Remove all states from beam.

thoth.adviser.boot module

A base class for implementing boot units.

class thoth.adviser.boot.Boot(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Boot base class implementation.

abstract run() → None

Main entry-point for boot units.

thoth.adviser.cli module

Thoth-adviser CLI.

thoth.adviser.context module

Pipeline context carried during annealing.

class thoth.adviser.context.Context(*, project: thoth.python.project.Project, graph: thoth.storages.graph.postgres.GraphDatabase, library_usage: Optional[Dict[str, Any]], limit: int, count: int, beam: thoth.adviser.beam.Beam, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, package_versions: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion] = NOTHING, dependencies: Dict[str, Dict[Tuple[str, str, str], Set[Tuple[str, str, str]]]] = NOTHING, dependents: Dict[str, Dict[Tuple[str, str, str], Set[Tuple[Tuple[str, str, str], Optional[str], Optional[str], Optional[str]]]]] = NOTHING, sources: Dict[str, thoth.python.source.Source] = NOTHING, iteration: int = 0, stack_info: Optional[List[Dict[str, Any]]] = NOTHING, accepted_final_states_count: int = 0, discarded_final_states_count: int = 0, accepted_states: List[Tuple[Tuple[float, int], thoth.adviser.state.State]] = NOTHING, accepted_states_counter: int = 0)

Bases: object

Context carried during adviser’s pipeline run.

It’s suitable to cache entries such as PackageVersion to optimize memory usage and optimize overhead needed - for example for parsing version strings (this is lazily pre-cached in PackageVersion).

accepted_final_states_count

beam

count

decision_type

dependencies

dependents

discarded_final_states_count

get_package_version(package_tuple: Tuple[str, str, str], *, graceful: bool = False) → Optional[thoth.python.package_version.PackageVersion]

Get the given package version registered to the context.

get_top_accepted_final_state() → Optional[thoth.adviser.state.State]

Get the best accepted final state so far computed by the resolution pipeline.

graph

is_adviser() → bool

Check if the current context refers to a dependency monkey run.

is_dependency_monkey() → bool

Check if the current context refers to a dependency monkey run.

iter_accepted_final_states() → Generator[[thoth.adviser.state.State, None], None]

Get accepted final states by resolution pipeline, states are not sorted.

iter_accepted_final_states_sorted(reverse: bool = True) → Generator[[thoth.adviser.state.State, None], None]

Get accepted final states by resolution pipeline sorted by score and their precedence.

iteration

library_usage

limit

package_versions

project

recommendation_type

register_accepted_final_state(state: thoth.adviser.state.State) → None

Register an accepted state by the resolution pipeline.

register_package_tuple(package_tuple: Tuple[str, str, str], *, develop: bool, dependent_tuple: Optional[Tuple[str, str, str]] = None, extras: Optional[List[str]] = None, os_name: Optional[str], os_version: Optional[str], python_version: Optional[str]) → thoth.python.package_version.PackageVersion

Register the given package tuple to pipeline context and return its package version representative.

register_package_version(package_version: thoth.python.package_version.PackageVersion) → bool

Register the given package version to the context.

sources

stack_info

thoth.adviser.dependency_monkey module

Routines for dependency monkey and its output handling.

class thoth.adviser.dependency_monkey.DependencyMonkey(*, resolver: thoth.adviser.resolver.Resolver, stack_output: str = '-', context: Optional[Dict[Any, Any]] = NOTHING, dry_run: bool = False, decision_type: thoth.adviser.enums.DecisionType = <DecisionType.ALL: 1>)

Bases: object

Dependency monkey creates stacks based on the configuration using ASA.

property beam

Get beam instance used in the resolver.

context

decision_type

dry_run

property predictor

Get predictor for the current dependency monkey configuration.

resolve(*, with_devel: bool = True) → thoth.adviser.dm_report.DependencyMonkeyReport

Perform simulated annealing and run dependency monkey on products.

resolver

stack_output

thoth.adviser.digests_fetcher module

Fetcher for fetching digests from the graph database.

class thoth.adviser.digests_fetcher.GraphDigestsFetcher(graph: thoth.storages.graph.postgres.GraphDatabase = NOTHING)

Bases: thoth.python.digests_fetcher.DigestsFetcherBase

Fetch digests from the graph database.

fetch_digests(package_name: str, package_version: str) → Dict[str, List[Dict[str, str]]]

Fetch digests for the given package in specified version, consider only enabled indexes.

graph

thoth.adviser.dm_report module

Routines for dependency monkey and its output handling.

class thoth.adviser.dm_report.DependencyMonkeyReport(skipped: int = 0, responses: List[Dict[str, Union[Dict[str, Any], str]]] = NOTHING)

Bases: object

Report produced by a Dependency Monkey run.

add_response(response: str, product: thoth.adviser.product.Product) → None

Add a new response to response listing.

skipped

to_dict() → Dict[str, Any]

Convert report to a dict representation suitable for serialization.

thoth.adviser.enums module

Enum types used in thoth-adviser code.

class thoth.adviser.enums.DecisionType

Bases: thoth.adviser.enums._ExtendedEnum

Type of decision used in Dependency Monkey for generating stacks.

ALL = 1

RANDOM = 2

class thoth.adviser.enums.Ecosystem

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1

class thoth.adviser.enums.PythonRecommendationOutput

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1

REQUIREMENTS = 2

class thoth.adviser.enums.RecommendationType

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3

STABLE = 1

TESTING = 2

thoth.adviser.exceptions module

Exception hierarchy used in the whole adviser implementation.

exception thoth.adviser.exceptions.AdviserException

Bases: Exception

A base adviser exception in adviser’s exception hierarchy.

exception thoth.adviser.exceptions.AdviserPipelineException

Bases: thoth.adviser.exceptions.AdviserException

A base class for implementing pipeline specific exceptions.

exception thoth.adviser.exceptions.AdviserRunException

Bases: thoth.adviser.exceptions.AdviserException

A base class for implementing exceptions occurred during an andviser run.

to_dict() → Optional[Dict[str, str]]

Convert adviser exception to a dict representation which is shown to the user.

exception thoth.adviser.exceptions.BootError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline boot unit fails unexpectedly.

exception thoth.adviser.exceptions.CannotProduceStack

Bases: thoth.adviser.exceptions.AdviserRunException

Raised if there was not produced any result.

to_dict() → Optional[Dict[str, str]]

Convert exception to a dict representation for a user.

exception thoth.adviser.exceptions.EagerStopPipeline

Bases: thoth.adviser.exceptions.AdviserPipelineException

Raised to signalize premature end of pipeline.

exception thoth.adviser.exceptions.InternalError

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on bugs in the code base.

exception thoth.adviser.exceptions.NoHistoryKept

Bases: thoth.adviser.exceptions.AdviserException

Raised if a user asks for history, but history was not kept (e.g. temperature function history in annealing).

exception thoth.adviser.exceptions.NotAcceptable

Bases: thoth.adviser.exceptions.AdviserException

An exception raised during stack generation when the given action would produce un-acceptable solution.

exception thoth.adviser.exceptions.NotFound

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if the given resource was not found.

exception thoth.adviser.exceptions.PipelineConfigurationError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when a wrong pipeline unit configuration supplied.

Or any error during configuration initialization.

exception thoth.adviser.exceptions.PipelineUnitError

Bases: thoth.adviser.exceptions.AdviserPipelineException

An exception raised when there is an error during pipeline run, unexpectedly.

exception thoth.adviser.exceptions.PipfileParseError

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on invalid Pipfile or Pipfile.lock.

exception thoth.adviser.exceptions.SieveError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline sieve unit fails unexpectedly.

exception thoth.adviser.exceptions.StepError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline step unit fails unexpectedly.

exception thoth.adviser.exceptions.StrideError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline stride unit fails unexpectedly.

exception thoth.adviser.exceptions.UnableLock

Bases: thoth.adviser.exceptions.AdviserException

Raised if it is unable to lock dependencies given the set of constraints.

exception thoth.adviser.exceptions.UnknownPipelineUnitError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when an unknown pipeline unit is requested.

exception thoth.adviser.exceptions.UnresolvedDependencies(*args: Any, unresolved: List[str])

Bases: thoth.adviser.exceptions.AdviserRunException

An exception raised if dependencies were not resolved and cannot produce stack.

to_dict() → Optional[Dict[str, str]]

Convert unresolved dependencies exception to the user.

unresolved

exception thoth.adviser.exceptions.UnsupportedConfiguration

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on unsupported configuration by recommendation engine.

exception thoth.adviser.exceptions.VersionIdentifierError

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if the given version identifier is not a semver identifier.

exception thoth.adviser.exceptions.WrapError

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline stride unit fails unexpectedly.

thoth.adviser.pipeline_builder module

Implementation of pipeline builder - create pipeline configuration based on the project and its library usage.

class thoth.adviser.pipeline_builder.PipelineBuilder

Bases: object

Builder responsible for creating pipeline configuration from the project and its library usage.

classmethod from_dict(dict_: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig

Instantiate pipeline configuration based on dictionary supplied.

classmethod get_adviser_pipeline_config(*, recommendation_type: thoth.adviser.enums.RecommendationType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig

Get adviser’s pipeline configuration.

classmethod get_dependency_monkey_pipeline_config(*, decision_type: thoth.adviser.enums.DecisionType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig

Get dependency-monkey’s pipeline configuration.

classmethod load(config: str) → thoth.adviser.pipeline_config.PipelineConfig

Load pipeline configuration from a file or a string.

class thoth.adviser.pipeline_builder.PipelineBuilderContext(*, graph: thoth.storages.graph.postgres.GraphDatabase = None, project: thoth.python.project.Project = None, library_usage: Optional[Dict[str, Any]] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING, boots_included: Set[Any] = NOTHING, sieves_included: Set[Any] = NOTHING, steps_included: Set[Any] = NOTHING, strides_included: Set[Any] = NOTHING, wraps_included: Set[Any] = NOTHING)

Bases: object

A context passed to units to determine if they want to participate in a pipeline.

add_unit(unit: thoth.adviser.unit.Unit) → None

Add the given unit to pipeline configuration.

property boots

Get all boots registered to this pipeline builder context.

decision_type

graph

is_adviser_pipeline() → bool

Check if the pipeline built is meant for adviser.

is_dependency_monkey_pipeline() → bool

Check if the pipeline built is meant for Dependency Monkey.

is_included(unit_class: type) → bool

Check if the given pipeline unit is already included in the pipeline configuration.

library_usage

project

recommendation_type

property sieves

Get all sieves registered to this pipeline builder context.

property steps

Get all steps registered to this pipeline builder context.

property strides

Get all strides registered to this pipeline builder context.

property wraps

Get all wraps registered to this pipeline builder context.

thoth.adviser.pipeline_config module

A base class for implementing sieves.

class thoth.adviser.pipeline_config.PipelineConfig(boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING)

Bases: object

A configuration of a pipeline for dependency-monkey and for adviser.

boots

call_post_run() → None

Call post-run method on all units registered in this configuration.

call_post_run_report(report: Union[Report, thoth.adviser.dm_report.DependencyMonkeyReport]) → None

Call post-run method when report is generated.

call_pre_run() → None

Call pre-run method on all units registered in this configuration.

iter_units() → Generator[[thoth.adviser.unit.Unit, None], None]

Iterate over units present in the configuration.

sieves

steps

strides

to_dict() → Dict[str, List[Dict[str, Any]]]

Return this pipeline configuration in a dict representation.

wraps

thoth.adviser.predictor module

A base class for implementing predictor.

class thoth.adviser.predictor.Predictor(*, keep_history=None)

Bases: object

A base class for implementing a predictor for resolver.

classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[[None, None], None]

Assign context to predictor.

property context

Get context in which the unit runs in.

finalize_state(state_id: int) → None

Finalizer called when the given state is about to be destructed by garbage collector.

Method suitable if predictor keeps internal state for states. Note that this method is not called for remaining states if the resolver terminates.

@param state_id: id of state that is about to be finalized

get_beam_key(state: thoth.adviser.state.State) → object

Retrieve a key that will be used to keep states in the beam.

The key returned should always uniquely sort two states, that is, there should not exist >>> state1 = State(score=1.0) # any score value >>> self.get_beam_key(state1) such as there exist state2 for which >>> state2 = State(score=1.0) # any score value >>> self.get_beam_key(state1) == self.get_beam_key(state2)

The simplest way how to differentiate keys of two states is to use resolver’s iteration, that is always unique for any state, and prioritizes states that have more resolved dependencies over the ones that have less resolved dependencies but the same score.

This method should not be treated as staticmethod or as classmethod - a predictor can construct key based on its own internal state.

keep_history

plot() → matplotlib.figure.Figure

Plot information about predictor.

post_run() → None

Post-run method run after the resolving has been done.

post_run_report(report: thoth.adviser.report.Report) → None

Post-run method run after the resolving has finished - this method is called only if resolving with a report.

The default operation is a noop, but a predictor can perform any post-run operations in this method. This method should not raise any exception.

pre_run() → None

Pre-initialize the predictor.

This method is called before any resolving with a freshly instantiated context. The default operation is a noop, but predictor can perform any initial setup in this method. This method should not raise any exception.

abstract run() → Tuple[thoth.adviser.state.State, Tuple[str, str, str]]

The main method used to run the predictor.

set_reward_signal(state: thoth.adviser.state.State, package_tuple: Tuple[str, str, str], reward: float) → None

Signalize the reward.

@param state: (child) state for which the reward signal is triggered @param package_tuple: Python package that was added to the state causing the reward @param reward: set to nan if the given state was not accepted a special value

of inf notifies about a new final state

thoth.adviser.product module

Representation of an advised stack.

class thoth.adviser.product.Product(project: thoth.python.project.Project, score: float, justification: List[Dict[str, str]], advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None)

Bases: object

A representation of an advised stack.

advised_runtime_environment

classmethod from_final_state(*, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product

Instantiate advised stack from final state produced by adviser’s pipeline.

justification

project

score

to_dict() → Dict[str, Any]

Convert this instance into a dictionary.

thoth.adviser.report module

A class for adviser’s report - output of an adviser run.

class thoth.adviser.report.Report(*, count: int, pipeline: thoth.adviser.pipeline_config.PipelineConfig, stack_info: Optional[List[Dict[str, Any]]] = None, heapq: List[Tuple[Tuple[float, int], thoth.adviser.product.Product]] = NOTHING, heapq_counter: int = 0)

Bases: object

A report stating output of an adviser run.

add_product(product: thoth.adviser.product.Product) → None

Add adviser pipeline product to report.

count

iter_products() → Generator[[thoth.adviser.product.Product, None], None]

Iterate over products stored in this report, respect their scores.

iter_products_sorted(reverse: bool = True) → Generator[[thoth.adviser.product.Product, None], None]

Iterate over products stored in this report, respect their scores.

pipeline

product_count() → int

Get number of products stored in the report.

set_stack_info(stack_info: List[Dict[str, Any]]) → None

Set stack information.

property stack_info

Retrieve stack info of this report.

to_dict() → Dict[str, Any]

Convert pipeline report to a dict representation.

thoth.adviser.resolver module

The main resolving algorithm working on top of states.

class thoth.adviser.resolver.Resolver(stop_resolving: bool = False, log_final_state_count: int = 500, log_final_state_top: bool = False, *, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, library_usage, graph: thoth.storages.graph.postgres.GraphDatabase, predictor: thoth.adviser.predictor.Predictor, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, limit: int = 10000, count: int = 3, beam_width=-1, limit_latest_versions=-1, beam: Optional[thoth.adviser.beam.Beam] = None, solver: Optional[thoth.adviser.solver.PythonPackageGraphSolver] = None, context: Optional[thoth.adviser.context.Context] = None, log_unresolved: Set[Tuple[str, str, str]] = NOTHING, log_unsolved: Set[str] = NOTHING, log_sieved: Set[str] = NOTHING, log_step_not_acceptable: Set[Tuple[str, str, str]] = NOTHING, log_no_intersected: Tuple[Tuple[str, str, str], str] = NOTHING)

Bases: object

Resolver for resolving software stacks using pipeline configuration and a predictor.

DEFAULT_BEAM_WIDTH = -1

DEFAULT_COUNT = 3

DEFAULT_LIMIT = 10000

DEFAULT_LIMIT_LATEST_VERSIONS = -1

DEFAULT_LOG_FINAL_STATE_COUNT = 500

DEFAULT_LOG_FINAL_STATE_TOP = False

property beam

Get beam for storing states.

beam_width

property context

Retrieve context bound to the current resolver.

count

decision_type

classmethod get_adviser_instance(*, predictor: thoth.adviser.predictor.Predictor, beam_width: Optional[int] = None, count: int = 3, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit: int = 10000, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, recommendation_type: thoth.adviser.enums.RecommendationType, pipeline_config: Union[thoth.adviser.pipeline_config.PipelineConfig, Dict[str, Any], None] = None) → thoth.adviser.resolver.Resolver

Get instance of resolver based on the project given to recommend software stacks.

classmethod get_dependency_monkey_instance(*, predictor: thoth.adviser.predictor.Predictor, beam_width: Optional[int] = None, count: int = 3, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, decision_type: thoth.adviser.enums.DecisionType, pipeline_config: Union[thoth.adviser.pipeline_config.PipelineConfig, Dict[str, Any], None] = None) → thoth.adviser.resolver.Resolver

Get instance of resolver based on the project given to run dependency monkey.

graph

library_usage

limit

limit_latest_versions

log_final_state_count

log_final_state_top

pipeline

predictor

project

recommendation_type

resolve(*, with_devel: bool = True) → thoth.adviser.report.Report

Resolve software stacks and return resolver report.

resolve_products(*, with_devel: bool = True) → Generator[[thoth.adviser.product.Product, None], None]

Resolve raw products as produced by this resolver pipeline.

property solver

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

stop_resolving

thoth.adviser.run module

Utilities which wrap adviser’s and dependency monkey’s and run them in a subprocess.

This is useful for a cluster deployment where we restrict resources allocated for an adviser or dependency monkey run. Annealing is run as a subprocess to the main process - if resources are exhausted (CPU time or memory allocated), the subprocess is killed (either by liveness probe or by OOM killer) and the main process can still produce an error message.

thoth.adviser.run.subprocess_run(resolver: Union[thoth.adviser.resolver.Resolver, thoth.adviser.dependency_monkey.DependencyMonkey], print_func: Callable[[float, Union[Dict[str, Any], List[Any]]], None], result_dict: Dict[str, Any], plot: Optional[str] = None) → int

Run the given function (partial annealing method) in a subprocess and output the produced report.

thoth.adviser.sieve module

A base class for implementing steps.

class thoth.adviser.sieve.Sieve(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Sieve base class implementation.

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

Main entry-point for sieves to filter and score packages.

thoth.adviser.solver module

Definition of package resolution based on precomputed data available in graph.

There are defined primitives required for offline resolution. This off-line resolution is done based on data aggregated in the graph database so thereis no need to perform resolving by actually installing Python packages (as this version resolution is dynamic in case of Python).

class thoth.adviser.solver.GraphReleasesFetcher(*, graph: thoth.storages.graph.postgres.GraphDatabase, runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = NOTHING)

Bases: thoth.solver.python.base.ReleasesFetcher

Fetch releases for packages from the graph database.

fetch_releases(package_name: str) → Tuple[str, List[Tuple[str, str]]]

Fetch releases for the given package name.

graph

runtime_environment

class thoth.adviser.solver.PackageVersionDependencyParser

Bases: thoth.solver.python.base.DependencyParser

Parse an instance of PackageVersion to Dependency object needed by solver.

parse(dependencies: List[thoth.python.package_version.PackageVersion]) → Generator[[packaging.requirements.Requirement, None], None]

Parse the given list of PackageVersion objects.

class thoth.adviser.solver.PythonGraphSolver(*, dependency_parser: thoth.adviser.solver.PackageVersionDependencyParser, releases_fetcher: thoth.adviser.solver.GraphReleasesFetcher)

Bases: thoth.solver.python.base.Solver

Solve Python dependencies based on data available in the graph database.

dependency_parser

releases_fetcher

class thoth.adviser.solver.PythonPackageGraphSolver(*, graph: thoth.storages.graph.postgres.GraphDatabase, runtime_environment: thoth.common.config.runtime_environment.RuntimeEnvironment = NOTHING, package_versions: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion] = NOTHING, sources: Dict[str, thoth.python.source.Source] = NOTHING, solver: thoth.adviser.solver.PythonGraphSolver = None)

Bases: object

A wrapper to manipulate with Python packages using pure PackageVersion object interface.

graph

runtime_environment

solve(dependencies: List[thoth.python.package_version.PackageVersion], graceful: bool = True) → Dict[str, List[thoth.python.package_version.PackageVersion]]

Solve the given dependencies and return object representation of packages.

property solver

Retrieve solver instance resolving using graph database.

thoth.adviser.state module

A state of not fully resolved software stack in adviser’s recommendations implementation.

class thoth.adviser.state.State(score: float = 0.0, iteration: int = 0, unresolved_dependencies=NOTHING, resolved_dependencies=NOTHING, parent: <module 'weakref' from '/home/user/.local/share/virtualenvs/adviser-F7Avhu3y/lib64/python3.6/weakref.py'> = None, *, advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None, justification: List[Dict[str, str]] = NOTHING)

Bases: object

Implementation of an adviser’s state in state space.

add_justification(justification: List[Dict[str, str]]) → None

Add new entries to the justification field.

add_resolved_dependency(package_tuple: Tuple[str, str, str]) → None

Add a resolved dependency into the state.

add_unresolved_dependency(package_tuple: Tuple[str, str, str]) → None

Add unresolved dependency into the state.

advised_runtime_environment

clone() → thoth.adviser.state.State

Return a swallow copy of this state that can be used as a next state.

classmethod from_direct_dependencies(direct_dependencies: Dict[str, List[thoth.python.package_version.PackageVersion]]) → thoth.adviser.state.State

Create an initial state out of direct dependencies.

get_first_unresolved_dependency() → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

get_random_first_unresolved_dependency() → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

get_random_unresolved_dependency(prefer_recent: bool = True) → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

is_final() → bool

Check if the given state is a final state.

iter_resolved_dependencies() → Generator[[Tuple[str, str, str], None], None]

Iterate over resolved dependencies.

iter_unresolved_dependencies() → Generator[[Tuple[str, str, str], None], None]

Iterate over unresolved dependencies.

iteration

justification

mark_dependency_resolved(package_tuple: Tuple[str, str, str]) → None

Mark the given dependency as resolved in the current state.

property parent

Retrieve parent to this state.

If None the state is top level state or parent is no longer maintained. Note the return value of None depends on actual gc runs.

remove_unresolved_dependency(package_tuple: Tuple[str, str, str]) → None

Remove the given unresolved dependency from state.

remove_unresolved_dependency_subtree(package_name: str) → None

Remove the whole dependency sub-tree from the state.

resolved_dependencies

score

set_unresolved_dependencies(dependencies: Dict[str, List[Tuple[str, str, str]]]) → None

Set unresolved dependencies - any unresolved dependencies will be overwritten.

to_dict() → Dict[str, Any]

Convert state to a dict representation.

unresolved_dependencies

thoth.adviser.step module

A base class for implementing steps.

class thoth.adviser.step.Step(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Step base class implementation.

Configuration option MUTLI_PACKAGE_RESOLUTION states whether this state should be run if package is resolved multiple times.

MULTI_PACKAGE_RESOLUTIONS = False

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

Main entry-point for steps to filter and score packages.

thoth.adviser.stride module

A base class for implementing strides.

class thoth.adviser.stride.Stride(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Stride base class implementation.

abstract run(state: thoth.adviser.state.State) → None

Main entry-point for strides.

thoth.adviser.unit module

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

class thoth.adviser.unit.Unit(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: object

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

CONFIGURATION_DEFAULT = {}

CONFIGURATION_SCHEMA = None

classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[[None, None], None]

Assign context to all units.

classmethod compute_expanded_configuration(configuration_dict: Optional[Dict[str, Any]]) → Dict[str, Any]

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

property configuration

Get configuration of instantiated pipeline unit.

property context

Get context in which the unit runs in.

classmethod get_aicoe_configuration(package_version: thoth.python.package_version.PackageVersion) → Optional[Dict[str, Any]]

Get AICoE specific configuration encoded in the AICoE index URL.

classmethod is_aicoe_release(package_version: thoth.python.package_version.PackageVersion) → bool

Check if the given package-version is AICoE release.

property name

Get name of this pipeline unit.

post_run() → None

Called after the resolution is finished.

This method should not raise any exception.

post_run_report(report: Union[Report, thoth.adviser.dm_report.DependencyMonkeyReport]) → None

Post-run method run after the resolving has finished - this method is called only if resolving with a report.

This method should not raise any exception.

pre_run() → None

Called before running any pipeline unit with context already assigned.

This method should not raise any exception.

classmethod should_include(builder_context: PipelineBuilderContext) → Optional[Dict[str, Any]]

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

to_dict() → Dict[str, Any]

Turn this pipeline step into its dictionary representation.

update_configuration(configuration_dict: Dict[str, Any]) → None

Set configuration for a pipeline unit.

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

thoth.adviser.utils module

Various helpers and utility functions.

thoth.adviser.utils.log_once(logger: logging.Logger, log_state: Set[object], log_state_key: object, msg: str, *args: object, level: int = 30, **kwargs: object) → None

Log the given message once.

thoth.adviser.utils.should_keep_history(value: Any) → bool

Check if the history should be kept.

Used as click’s converter. If not set explicitly during invocation, check environment variable to turn off history tracking. The default value of `value’ is None which triggers checks in environment variables.

thoth.adviser.wrap module

A base class for implementing wrap units.

class thoth.adviser.wrap.Wrap(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Wrap base class implementation.

abstract run(state: thoth.adviser.state.State) → None

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

Module contents

Thoth’s adviser for recommending Python stacks.

class thoth.adviser.Beam(width: Optional[int] = None, states: List[Tuple[Tuple[float, int], thoth.adviser.state.State]] = NOTHING, states_idx: Dict[int, int] = NOTHING, last_added: Optional[thoth.adviser.state.State] = None, top: Optional[thoth.adviser.state.State] = None, *, keep_history=None, beam_history: List[Tuple[int, Optional[float]]] = NOTHING)

Bases: object

Beam implementation.

The implementation of beam respecting beam width: https://en.wikipedia.org/wiki/Beam_search

Beam is internally implemented on top of a min-heap queue to optimize inserts and respect beam width in O(log(N)).

The most frequent operation performed on the beam is not always the same - it depends on the beam width and number of states generated. If number of states generated is more than the width of the beam, its reasonable to optimize insertions into the beam with checks on beam width.

If number of states generated is smaller than the beam width, the beam could be optimized for removal of states.

As the first scenario is seen in a real-world deployment, the beam uses min-heapq to keep addition to the beam with beam_width checks in O(log(N)) and removals of the states in O(log(N)). To satisfy removals in O(log(N)), the beam maintains a dictionary mapping a state to its index in the beam.

add_state(key: object, state: thoth.adviser.state.State) → None

Add state to the internal state listing (do it in O(log(N)) time.

get(idx: int) → thoth.adviser.state.State

Get i-th element from the beam (constant time), keep it in the beam.

This method is suitable for random state indexing in the beam (like in case of adaptive simulated annealing). The actual index is not into a sorted array and has no special meaning assigned - beam under the hood uses min-heapq (as of now), but the index used is not guaranteed to point to a heap-like data structure.

get_by_id(state_id: int) → thoth.adviser.state.State

Get a state by its id.

get_last() → Optional[thoth.adviser.state.State]

Get state that was added in the previous resolution round.

get_random() → thoth.adviser.state.State

Get a random state from beam.

iter_states() → Generator[[thoth.adviser.state.State, None], None]

Iterate over states, do not respect their score in order of iteration.

iter_states_sorted(reverse: bool = True) → Generator[[thoth.adviser.state.State, None], None]

Iterate over sorted states.

keep_history

new_iteration() → None

Called once a new iteration is done in resolver.

Used to keep track of beam history and to keep track of states added.

plot() → matplotlib.figure.Figure

Plot temperature history of adaptive simulated annealing.

pop(idx: Optional[int] = None) → thoth.adviser.state.State

Pop i-th element from the beam and remove it from the beam (this is actually toppop).

If index is not provided, pop one of the largest elements kept in the beam.

If top_state is pre-cached or idx is explicitly set, this operation is done in O(log(N)), O(N) otherwise.

remove(state: thoth.adviser.state.State) → None

Remove the given state from beam.

reset() → None

Recompute internal state of the beam if any changes to the score of states were made.

property size

Get the current size of beam.

top() → thoth.adviser.state.State

Return the highest rated state as kept in the beam.

This operation is done in O(N) if top is not pre-cached, O(1) otherwise.

width

wipe() → None

Remove all states from beam.

class thoth.adviser.Boot(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Boot base class implementation.

abstract run() → None

Main entry-point for boot units.

class thoth.adviser.Context(*, project: thoth.python.project.Project, graph: thoth.storages.graph.postgres.GraphDatabase, library_usage: Optional[Dict[str, Any]], limit: int, count: int, beam: thoth.adviser.beam.Beam, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, package_versions: Dict[Tuple[str, str, str], thoth.python.package_version.PackageVersion] = NOTHING, dependencies: Dict[str, Dict[Tuple[str, str, str], Set[Tuple[str, str, str]]]] = NOTHING, dependents: Dict[str, Dict[Tuple[str, str, str], Set[Tuple[Tuple[str, str, str], Optional[str], Optional[str], Optional[str]]]]] = NOTHING, sources: Dict[str, thoth.python.source.Source] = NOTHING, iteration: int = 0, stack_info: Optional[List[Dict[str, Any]]] = NOTHING, accepted_final_states_count: int = 0, discarded_final_states_count: int = 0, accepted_states: List[Tuple[Tuple[float, int], thoth.adviser.state.State]] = NOTHING, accepted_states_counter: int = 0)

Bases: object

Context carried during adviser’s pipeline run.

It’s suitable to cache entries such as PackageVersion to optimize memory usage and optimize overhead needed - for example for parsing version strings (this is lazily pre-cached in PackageVersion).

accepted_final_states_count

beam

count

decision_type

dependencies

dependents

discarded_final_states_count

get_package_version(package_tuple: Tuple[str, str, str], *, graceful: bool = False) → Optional[thoth.python.package_version.PackageVersion]

Get the given package version registered to the context.

get_top_accepted_final_state() → Optional[thoth.adviser.state.State]

Get the best accepted final state so far computed by the resolution pipeline.

graph

is_adviser() → bool

Check if the current context refers to a dependency monkey run.

is_dependency_monkey() → bool

Check if the current context refers to a dependency monkey run.

iter_accepted_final_states() → Generator[[thoth.adviser.state.State, None], None]

Get accepted final states by resolution pipeline, states are not sorted.

iter_accepted_final_states_sorted(reverse: bool = True) → Generator[[thoth.adviser.state.State, None], None]

Get accepted final states by resolution pipeline sorted by score and their precedence.

iteration

library_usage

limit

package_versions

project

recommendation_type

register_accepted_final_state(state: thoth.adviser.state.State) → None

Register an accepted state by the resolution pipeline.

register_package_tuple(package_tuple: Tuple[str, str, str], *, develop: bool, dependent_tuple: Optional[Tuple[str, str, str]] = None, extras: Optional[List[str]] = None, os_name: Optional[str], os_version: Optional[str], python_version: Optional[str]) → thoth.python.package_version.PackageVersion

Register the given package tuple to pipeline context and return its package version representative.

register_package_version(package_version: thoth.python.package_version.PackageVersion) → bool

Register the given package version to the context.

sources

stack_info

class thoth.adviser.DecisionType

Bases: thoth.adviser.enums._ExtendedEnum

Type of decision used in Dependency Monkey for generating stacks.

ALL = 1

RANDOM = 2

class thoth.adviser.DependencyMonkey(*, resolver: thoth.adviser.resolver.Resolver, stack_output: str = '-', context: Optional[Dict[Any, Any]] = NOTHING, dry_run: bool = False, decision_type: thoth.adviser.enums.DecisionType = <DecisionType.ALL: 1>)

Bases: object

Dependency monkey creates stacks based on the configuration using ASA.

property beam

Get beam instance used in the resolver.

context

decision_type

dry_run

property predictor

Get predictor for the current dependency monkey configuration.

resolve(*, with_devel: bool = True) → thoth.adviser.dm_report.DependencyMonkeyReport

Perform simulated annealing and run dependency monkey on products.

resolver

stack_output

class thoth.adviser.DependencyMonkeyReport(skipped: int = 0, responses: List[Dict[str, Union[Dict[str, Any], str]]] = NOTHING)

Bases: object

Report produced by a Dependency Monkey run.

add_response(response: str, product: thoth.adviser.product.Product) → None

Add a new response to response listing.

skipped

to_dict() → Dict[str, Any]

Convert report to a dict representation suitable for serialization.

class thoth.adviser.Ecosystem

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1

class thoth.adviser.PipelineBuilder

Bases: object

Builder responsible for creating pipeline configuration from the project and its library usage.

classmethod from_dict(dict_: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig

Instantiate pipeline configuration based on dictionary supplied.

classmethod get_adviser_pipeline_config(*, recommendation_type: thoth.adviser.enums.RecommendationType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig

Get adviser’s pipeline configuration.

classmethod get_dependency_monkey_pipeline_config(*, decision_type: thoth.adviser.enums.DecisionType, graph: thoth.storages.graph.postgres.GraphDatabase, project: thoth.python.project.Project, library_usage: Optional[Dict[str, Any]]) → thoth.adviser.pipeline_config.PipelineConfig

Get dependency-monkey’s pipeline configuration.

classmethod load(config: str) → thoth.adviser.pipeline_config.PipelineConfig

Load pipeline configuration from a file or a string.

class thoth.adviser.PipelineConfig(boots: List[thoth.adviser.boot.Boot] = NOTHING, sieves: List[thoth.adviser.sieve.Sieve] = NOTHING, steps: List[thoth.adviser.step.Step] = NOTHING, strides: List[thoth.adviser.stride.Stride] = NOTHING, wraps: List[thoth.adviser.wrap.Wrap] = NOTHING)

Bases: object

A configuration of a pipeline for dependency-monkey and for adviser.

boots

call_post_run() → None

Call post-run method on all units registered in this configuration.

call_post_run_report(report: Union[Report, thoth.adviser.dm_report.DependencyMonkeyReport]) → None

Call post-run method when report is generated.

call_pre_run() → None

Call pre-run method on all units registered in this configuration.

iter_units() → Generator[[thoth.adviser.unit.Unit, None], None]

Iterate over units present in the configuration.

sieves

steps

strides

to_dict() → Dict[str, List[Dict[str, Any]]]

Return this pipeline configuration in a dict representation.

wraps

class thoth.adviser.Predictor(*, keep_history=None)

Bases: object

A base class for implementing a predictor for resolver.

classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[[None, None], None]

Assign context to predictor.

property context

Get context in which the unit runs in.

finalize_state(state_id: int) → None

Finalizer called when the given state is about to be destructed by garbage collector.

Method suitable if predictor keeps internal state for states. Note that this method is not called for remaining states if the resolver terminates.

@param state_id: id of state that is about to be finalized

get_beam_key(state: thoth.adviser.state.State) → object

Retrieve a key that will be used to keep states in the beam.

The key returned should always uniquely sort two states, that is, there should not exist >>> state1 = State(score=1.0) # any score value >>> self.get_beam_key(state1) such as there exist state2 for which >>> state2 = State(score=1.0) # any score value >>> self.get_beam_key(state1) == self.get_beam_key(state2)

The simplest way how to differentiate keys of two states is to use resolver’s iteration, that is always unique for any state, and prioritizes states that have more resolved dependencies over the ones that have less resolved dependencies but the same score.

This method should not be treated as staticmethod or as classmethod - a predictor can construct key based on its own internal state.

keep_history

plot() → matplotlib.figure.Figure

Plot information about predictor.

post_run() → None

Post-run method run after the resolving has been done.

post_run_report(report: thoth.adviser.report.Report) → None

Post-run method run after the resolving has finished - this method is called only if resolving with a report.

The default operation is a noop, but a predictor can perform any post-run operations in this method. This method should not raise any exception.

pre_run() → None

Pre-initialize the predictor.

This method is called before any resolving with a freshly instantiated context. The default operation is a noop, but predictor can perform any initial setup in this method. This method should not raise any exception.

abstract run() → Tuple[thoth.adviser.state.State, Tuple[str, str, str]]

The main method used to run the predictor.

set_reward_signal(state: thoth.adviser.state.State, package_tuple: Tuple[str, str, str], reward: float) → None

Signalize the reward.

@param state: (child) state for which the reward signal is triggered @param package_tuple: Python package that was added to the state causing the reward @param reward: set to nan if the given state was not accepted a special value

of inf notifies about a new final state

class thoth.adviser.Product(project: thoth.python.project.Project, score: float, justification: List[Dict[str, str]], advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None)

Bases: object

A representation of an advised stack.

advised_runtime_environment

classmethod from_final_state(*, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product

Instantiate advised stack from final state produced by adviser’s pipeline.

justification

project

score

to_dict() → Dict[str, Any]

Convert this instance into a dictionary.

class thoth.adviser.PythonRecommendationOutput

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1

REQUIREMENTS = 2

class thoth.adviser.RecommendationType

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3

STABLE = 1

TESTING = 2

class thoth.adviser.Report(*, count: int, pipeline: thoth.adviser.pipeline_config.PipelineConfig, stack_info: Optional[List[Dict[str, Any]]] = None, heapq: List[Tuple[Tuple[float, int], thoth.adviser.product.Product]] = NOTHING, heapq_counter: int = 0)

Bases: object

A report stating output of an adviser run.

add_product(product: thoth.adviser.product.Product) → None

Add adviser pipeline product to report.

count

iter_products() → Generator[[thoth.adviser.product.Product, None], None]

Iterate over products stored in this report, respect their scores.

iter_products_sorted(reverse: bool = True) → Generator[[thoth.adviser.product.Product, None], None]

Iterate over products stored in this report, respect their scores.

pipeline

product_count() → int

Get number of products stored in the report.

set_stack_info(stack_info: List[Dict[str, Any]]) → None

Set stack information.

property stack_info

Retrieve stack info of this report.

to_dict() → Dict[str, Any]

Convert pipeline report to a dict representation.

class thoth.adviser.Resolver(stop_resolving: bool = False, log_final_state_count: int = 500, log_final_state_top: bool = False, *, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, library_usage, graph: thoth.storages.graph.postgres.GraphDatabase, predictor: thoth.adviser.predictor.Predictor, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, limit: int = 10000, count: int = 3, beam_width=-1, limit_latest_versions=-1, beam: Optional[thoth.adviser.beam.Beam] = None, solver: Optional[thoth.adviser.solver.PythonPackageGraphSolver] = None, context: Optional[thoth.adviser.context.Context] = None, log_unresolved: Set[Tuple[str, str, str]] = NOTHING, log_unsolved: Set[str] = NOTHING, log_sieved: Set[str] = NOTHING, log_step_not_acceptable: Set[Tuple[str, str, str]] = NOTHING, log_no_intersected: Tuple[Tuple[str, str, str], str] = NOTHING)

Bases: object

Resolver for resolving software stacks using pipeline configuration and a predictor.

DEFAULT_BEAM_WIDTH = -1

DEFAULT_COUNT = 3

DEFAULT_LIMIT = 10000

DEFAULT_LIMIT_LATEST_VERSIONS = -1

DEFAULT_LOG_FINAL_STATE_COUNT = 500

DEFAULT_LOG_FINAL_STATE_TOP = False

property beam

Get beam for storing states.

beam_width

property context

Retrieve context bound to the current resolver.

count

decision_type

classmethod get_adviser_instance(*, predictor: thoth.adviser.predictor.Predictor, beam_width: Optional[int] = None, count: int = 3, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit: int = 10000, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, recommendation_type: thoth.adviser.enums.RecommendationType, pipeline_config: Union[thoth.adviser.pipeline_config.PipelineConfig, Dict[str, Any], None] = None) → thoth.adviser.resolver.Resolver

Get instance of resolver based on the project given to recommend software stacks.

classmethod get_dependency_monkey_instance(*, predictor: thoth.adviser.predictor.Predictor, beam_width: Optional[int] = None, count: int = 3, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, library_usage: Optional[Dict[str, Any]] = None, limit_latest_versions: Optional[int] = -1, project: thoth.python.project.Project, decision_type: thoth.adviser.enums.DecisionType, pipeline_config: Union[thoth.adviser.pipeline_config.PipelineConfig, Dict[str, Any], None] = None) → thoth.adviser.resolver.Resolver

Get instance of resolver based on the project given to run dependency monkey.

graph

library_usage

limit

limit_latest_versions

log_final_state_count

log_final_state_top

pipeline

predictor

project

recommendation_type

resolve(*, with_devel: bool = True) → thoth.adviser.report.Report

Resolve software stacks and return resolver report.

resolve_products(*, with_devel: bool = True) → Generator[[thoth.adviser.product.Product, None], None]

Resolve raw products as produced by this resolver pipeline.

property solver

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

stop_resolving

class thoth.adviser.Sieve(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Sieve base class implementation.

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

Main entry-point for sieves to filter and score packages.

class thoth.adviser.State(score: float = 0.0, iteration: int = 0, unresolved_dependencies=NOTHING, resolved_dependencies=NOTHING, parent: <module 'weakref' from '/home/user/.local/share/virtualenvs/adviser-F7Avhu3y/lib64/python3.6/weakref.py'> = None, *, advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None, justification: List[Dict[str, str]] = NOTHING)

Bases: object

Implementation of an adviser’s state in state space.

add_justification(justification: List[Dict[str, str]]) → None

Add new entries to the justification field.

add_resolved_dependency(package_tuple: Tuple[str, str, str]) → None

Add a resolved dependency into the state.

add_unresolved_dependency(package_tuple: Tuple[str, str, str]) → None

Add unresolved dependency into the state.

advised_runtime_environment

clone() → thoth.adviser.state.State

Return a swallow copy of this state that can be used as a next state.

classmethod from_direct_dependencies(direct_dependencies: Dict[str, List[thoth.python.package_version.PackageVersion]]) → thoth.adviser.state.State

Create an initial state out of direct dependencies.

get_first_unresolved_dependency() → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

get_random_first_unresolved_dependency() → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

get_random_unresolved_dependency(prefer_recent: bool = True) → Tuple[str, str, str]

Get a very first unresolved dependency tuple.

is_final() → bool

Check if the given state is a final state.

iter_resolved_dependencies() → Generator[[Tuple[str, str, str], None], None]

Iterate over resolved dependencies.

iter_unresolved_dependencies() → Generator[[Tuple[str, str, str], None], None]

Iterate over unresolved dependencies.

iteration

justification

mark_dependency_resolved(package_tuple: Tuple[str, str, str]) → None

Mark the given dependency as resolved in the current state.

property parent

Retrieve parent to this state.

If None the state is top level state or parent is no longer maintained. Note the return value of None depends on actual gc runs.

remove_unresolved_dependency(package_tuple: Tuple[str, str, str]) → None

Remove the given unresolved dependency from state.

remove_unresolved_dependency_subtree(package_name: str) → None

Remove the whole dependency sub-tree from the state.

resolved_dependencies

score

set_unresolved_dependencies(dependencies: Dict[str, List[Tuple[str, str, str]]]) → None

Set unresolved dependencies - any unresolved dependencies will be overwritten.

to_dict() → Dict[str, Any]

Convert state to a dict representation.

unresolved_dependencies

class thoth.adviser.Step(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Step base class implementation.

Configuration option MUTLI_PACKAGE_RESOLUTION states whether this state should be run if package is resolved multiple times.

MULTI_PACKAGE_RESOLUTIONS = False

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

Main entry-point for steps to filter and score packages.

class thoth.adviser.Stride(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Stride base class implementation.

abstract run(state: thoth.adviser.state.State) → None

Main entry-point for strides.

class thoth.adviser.Unit(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: object

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

CONFIGURATION_DEFAULT = {}

CONFIGURATION_SCHEMA = None

classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[[None, None], None]

Assign context to all units.

classmethod compute_expanded_configuration(configuration_dict: Optional[Dict[str, Any]]) → Dict[str, Any]

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

property configuration

Get configuration of instantiated pipeline unit.

property context

Get context in which the unit runs in.

classmethod get_aicoe_configuration(package_version: thoth.python.package_version.PackageVersion) → Optional[Dict[str, Any]]

Get AICoE specific configuration encoded in the AICoE index URL.

classmethod is_aicoe_release(package_version: thoth.python.package_version.PackageVersion) → bool

Check if the given package-version is AICoE release.

property name

Get name of this pipeline unit.

post_run() → None

Called after the resolution is finished.

This method should not raise any exception.

post_run_report(report: Union[Report, thoth.adviser.dm_report.DependencyMonkeyReport]) → None

Post-run method run after the resolving has finished - this method is called only if resolving with a report.

This method should not raise any exception.

pre_run() → None

Called before running any pipeline unit with context already assigned.

This method should not raise any exception.

classmethod should_include(builder_context: PipelineBuilderContext) → Optional[Dict[str, Any]]

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

to_dict() → Dict[str, Any]

Turn this pipeline step into its dictionary representation.

update_configuration(configuration_dict: Dict[str, Any]) → None

Set configuration for a pipeline unit.

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

class thoth.adviser.Wrap(*, configuration: Dict[str, Any] = NOTHING, name: str = None)

Bases: thoth.adviser.unit.Unit

Wrap base class implementation.

abstract run(state: thoth.adviser.state.State) → None

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