thoth.adviser package

Subpackages

Submodules

thoth.adviser.beam module

Implementation of Beam for beam searching part of adviser.

class thoth.adviser.beam.Beam(width: Optional[int] = None, *, keep_history: Any = None, beam_history: List[Tuple[int, Optional[float]]] = NOTHING)[source]

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(state: thoth.adviser.state.State) → None[source]

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

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

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_last() → Optional[thoth.adviser.state.State][source]

Get state that was added in the previous resolution round.

get_random() → thoth.adviser.state.State[source]

Get a random state from beam.

iter_states() → List[thoth.adviser.state.State][source]

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][source]

Iterate over sorted states.

keep_history
max() → thoth.adviser.state.State[source]

Return the highest rated state as kept in the beam.

new_iteration() → None[source]

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[source]

Plot temperature history of adaptive simulated annealing.

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

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

If index is not provided, pop the largest item kept in the beam.

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

Remove the given state from beam.

property size

Get the current size of beam.

width
wipe() → None[source]

Remove all states from beam.

thoth.adviser.boot module

A base class for implementing boot units.

class thoth.adviser.boot.Boot(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Boot base class implementation.

static is_boot_unit_type() → bool[source]

Check if this unit is of type boot.

abstract run() → None[source]

Run 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, labels: Dict[str, str], 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, prescription: Optional[Prescription] = None, cli_parameters: Dict[str, Any] = NOTHING, stack_info: 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)[source]

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
cli_parameters
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][source]

Get the given package version registered to the context.

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

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

graph
is_adviser() → bool[source]

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

is_dependency_monkey() → bool[source]

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

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

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][source]

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

iteration
labels
library_usage
limit
package_versions
prescription
project
recommendation_type
register_accepted_final_state(state: thoth.adviser.state.State) → None[source]

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[source]

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[source]

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>)[source]

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
plot() → matplotlib.figure.Figure[source]

Plot info from Dependency Monkey run.

property predictor

Get predictor for the current dependency monkey configuration.

resolve(*, with_devel: bool = True, user_stack_scoring: bool = False) → thoth.adviser.dm_report.DependencyMonkeyReport[source]

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)[source]

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]]][source]

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)[source]

Bases: object

Report produced by a Dependency Monkey run.

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

Add a new response to response listing.

skipped
to_dict() → Dict[str, Any][source]

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[source]

Bases: thoth.adviser.enums._ExtendedEnum

Type of decision used in Dependency Monkey for generating stacks.

ALL = 1
RANDOM = 2
class thoth.adviser.enums.Ecosystem[source]

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1
class thoth.adviser.enums.PythonRecommendationOutput[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1
REQUIREMENTS = 2
class thoth.adviser.enums.RecommendationType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3
PERFORMANCE = 4
SECURITY = 5
STABLE = 1
TESTING = 2

thoth.adviser.exceptions module

Exception hierarchy used in the whole adviser implementation.

exception thoth.adviser.exceptions.AdviserException[source]

Bases: Exception

A base adviser exception in adviser’s exception hierarchy.

exception thoth.adviser.exceptions.AdviserPipelineException[source]

Bases: thoth.adviser.exceptions.AdviserException

A base class for implementing pipeline specific exceptions.

exception thoth.adviser.exceptions.AdviserRunException[source]

Bases: thoth.adviser.exceptions.AdviserException

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

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

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

exception thoth.adviser.exceptions.BootError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline boot unit fails unexpectedly.

exception thoth.adviser.exceptions.CannotProduceStack(*args: Any, stack_info: List[Dict[str, Any]])[source]

Bases: thoth.adviser.exceptions.AdviserRunException

Raised if there was not produced any result.

stack_info
to_dict() → Optional[Dict[str, Any]][source]

Convert exception to a dict representation for a user.

exception thoth.adviser.exceptions.EagerStopPipeline[source]

Bases: thoth.adviser.exceptions.AdviserPipelineException

Raised to signalize premature end of pipeline.

exception thoth.adviser.exceptions.InternalError[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on bugs in the code base.

exception thoth.adviser.exceptions.NoHistoryKept[source]

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[source]

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[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised if the given resource was not found.

exception thoth.adviser.exceptions.PipelineConfigurationError[source]

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.PipelineUnitConfigurationSchemaError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline unit configuration does not match schema declared.

exception thoth.adviser.exceptions.PipelineUnitError[source]

Bases: thoth.adviser.exceptions.AdviserPipelineException

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

exception thoth.adviser.exceptions.PipfileParseError[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on invalid Pipfile or Pipfile.lock.

exception thoth.adviser.exceptions.PrescriptionDuplicateUnitNameError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when multiple prescription units share name.

exception thoth.adviser.exceptions.PrescriptionSchemaError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when prescription schema is not valid.

exception thoth.adviser.exceptions.SieveError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline sieve unit fails unexpectedly.

exception thoth.adviser.exceptions.SkipPackage[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised during stack generation when the given package should be skipped in the stack.

exception thoth.adviser.exceptions.StepError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline step unit fails unexpectedly.

exception thoth.adviser.exceptions.StrideError[source]

Bases: thoth.adviser.exceptions.PipelineUnitError

An exception raised when pipeline stride unit fails unexpectedly.

exception thoth.adviser.exceptions.UnableLock[source]

Bases: thoth.adviser.exceptions.AdviserException

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

exception thoth.adviser.exceptions.UnknownPipelineUnitError[source]

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], stack_info: List[Dict[str, Any]])[source]

Bases: thoth.adviser.exceptions.AdviserRunException

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

stack_info
to_dict() → Optional[Dict[str, Any]][source]

Convert unresolved dependencies exception to the user.

unresolved
exception thoth.adviser.exceptions.UnsupportedConfiguration[source]

Bases: thoth.adviser.exceptions.AdviserException

An exception raised on unsupported configuration by recommendation engine.

exception thoth.adviser.exceptions.UserLockFileError[source]

Bases: thoth.adviser.exceptions.AdviserRunException

An exception raised when the supplied user stack has issues.

exception thoth.adviser.exceptions.VersionIdentifierError[source]

Bases: thoth.adviser.exceptions.AdviserException

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

exception thoth.adviser.exceptions.WrapError[source]

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[source]

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[source]

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, labels: Dict[str, str], library_usage: Optional[Dict[str, Any]], prescription: Optional[Prescription], cli_parameters: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig[source]

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, labels: Dict[str, str], library_usage: Optional[Dict[str, Any]], prescription: Optional[Prescription], cli_parameters: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get dependency-monkey’s pipeline configuration.

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

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, labels: Dict[str, str] = NOTHING, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, cli_parameters: Dict[str, Any] = NOTHING, prescription: Optional[Prescription] = None, iteration: int = 0, boots: Dict[Optional[str], List[BootType]] = NOTHING, pseudonyms: Dict[str, List[PseudonymType]] = NOTHING, sieves: Dict[Optional[str], List[SieveType]] = NOTHING, steps: Dict[Optional[str], List[StepType]] = NOTHING, strides: Dict[Optional[str], List[StrideType]] = NOTHING, wraps: Dict[Optional[str], List[WrapType]] = NOTHING, boots_included: Dict[str, List[BootType]] = NOTHING, pseudonyms_included: Dict[str, List[PseudonymType]] = NOTHING, sieves_included: Dict[str, List[SieveType]] = NOTHING, steps_included: Dict[str, List[StepType]] = NOTHING, strides_included: Dict[str, List[StrideType]] = NOTHING, wraps_included: Dict[str, List[WrapType]] = NOTHING)[source]

Bases: object

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

add_unit(unit: UnitType) → None[source]

Add the given unit to pipeline configuration.

property boots

Get all boots registered to this pipeline builder context.

property boots_dict

Get boots as a dictionary mapping.

cli_parameters
decision_type
get_included_boot_names() → Generator[str, None, None][source]

Get names of included boots.

get_included_boots(boot_class: Type[UnitType]) → Generator[BootType, None, None][source]

Get included boots of the provided boot class.

get_included_pseudonym_names() → Generator[str, None, None][source]

Get names of included pseudonyms.

get_included_pseudonyms(pseudonym_class: Type[PseudonymType]) → Generator[PseudonymType, None, None][source]

Get included sieves of the provided sieve class.

get_included_sieve_names() → Generator[str, None, None][source]

Get names of included sieves.

get_included_sieves(sieve_class: Type[SieveType]) → Generator[SieveType, None, None][source]

Get included sieves of the provided sieve class.

get_included_step_names() → Generator[str, None, None][source]

Get names of included steps.

get_included_steps(step_class: Type[StepType]) → Generator[StepType, None, None][source]

Get included steps of the provided step class.

get_included_stride_names() → Generator[str, None, None][source]

Get names of included strides.

get_included_strides(stride_class: Type[StrideType]) → Generator[StrideType, None, None][source]

Get included strides of the provided stride class.

get_included_wrap_names() → Generator[str, None, None][source]

Get names of included wraps.

get_included_wraps(wrap_class: Type[WrapType]) → Generator[WrapType, None, None][source]

Get included wraps of the provided wrap class.

graph
is_adviser_pipeline() → bool[source]

Check if the pipeline built is meant for adviser.

is_dependency_monkey_pipeline() → bool[source]

Check if the pipeline built is meant for Dependency Monkey.

is_included(unit_class: Type[UnitType]) → bool[source]

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

iteration
labels
library_usage
prescription
project
property pseudonyms

Get all pseudonyms registered to this pipeline builder context.

property pseudonyms_dict

Get pseudonyms as a dictionary mapping.

recommendation_type
property sieves

Get all sieves registered to this pipeline builder context.

property sieves_dict

Get sieves as a dictionary mapping.

property steps

Get all steps registered to this pipeline builder context.

property steps_dict

Get steps as a dictionary mapping.

property strides

Get all strides registered to this pipeline builder context.

property strides_dict

Get strides as a dictionary mapping.

property wraps

Get all wraps registered to this pipeline builder context.

property wraps_dict

Get wraps as a dictionary mapping.

thoth.adviser.pipeline_config module

A base class for implementing sieves.

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

Bases: object

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

property boots

Get all boots.

property boots_dict

Get boots as a dictionary mapping.

call_post_run() → None[source]

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

call_post_run_report(report)[source]

Call post-run method when report is generated.

call_pre_run() → None[source]

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

iter_units()[source]

Iterate over units present in the configuration.

iter_units_reversed()[source]

Iterate over units present in the configuration in a reversed order.

property pseudonyms

Get all pseudonyms.

property pseudonyms_dict

Get pseudonyms as a dictionary mapping.

property sieves

Get all sieves.

property sieves_dict

Get sieves as a dictionary mapping.

property steps

Get all steps.

property steps_dict

Get steps as a dictionary mapping.

property strides

Get all strides.

property strides_dict

Get strides as a dictionary mapping.

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

Return this pipeline configuration in a dict representation.

property wraps

Get all wraps.

property wraps_dict

Get wraps as a dictionary mapping.

thoth.adviser.predictor module

A base class for implementing predictor.

class thoth.adviser.predictor.Predictor(*, keep_history: Any = None)[source]

Bases: object

A base class for implementing a predictor for resolver.

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

Assign context to predictor.

property context

Get context in which the unit runs in.

finalize_state(state_id: int) → None[source]

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

keep_history
classmethod obtain_default_configuration(config_option_name: str) → Any[source]

Get the default value assigned to the given configuration option of a predictor instance.

plot() → matplotlib.figure.Figure[source]

Plot information about predictor.

post_run() → None[source]

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

post_run_report(report: thoth.adviser.report.Report) → None[source]

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[source]

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]][source]

Run 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[source]

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, advised_manifest_changes: List[List[Dict[str, Any]]] = NOTHING)[source]

Bases: object

A representation of an advised stack.

advised_manifest_changes
advised_runtime_environment
classmethod from_final_state(*, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product[source]

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

justification
project
score
to_dict() → Dict[str, Any][source]

Convert this instance into a dictionary.

thoth.adviser.pseudonym module

A base class for implementing package pseudonyms.

class thoth.adviser.pseudonym.Pseudonym(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Pseudonym base class implementation.

CONFIGURATION_SCHEMA = <Schema({'package_name': <class 'str'>}, extra=PREVENT_EXTRA, required=False) object>
static is_pseudonym_unit_type() → bool[source]

Check if this unit is of type pseudonym.

abstract 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.

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, resolver_iterations: int = 0, accepted_final_states_count: int = 0, discarded_final_states_count: int = 0, stack_info: Optional[List[Dict[str, Any]]] = None, heapq: List[Tuple[Tuple[float, int], thoth.adviser.product.Product]] = NOTHING, heapq_counter: int = 0)[source]

Bases: object

A report stating output of an adviser run.

accepted_final_states_count
add_product(product: thoth.adviser.product.Product) → bool[source]

Add adviser pipeline product to report.

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

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

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

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

pipeline
product_count() → int[source]

Get number of products stored in the report.

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

Set stack information.

property stack_info

Retrieve stack info of this report.

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

Convert pipeline report to a dict representation.

will_add_product_from_state(score: float) → bool[source]

Check if a product with the given score would be added.

thoth.adviser.resolver module

The main resolving algorithm working on top of states.

class thoth.adviser.resolver.Resolver(*, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, library_usage: Any, graph: thoth.storages.graph.postgres.GraphDatabase, predictor: thoth.adviser.predictor.Predictor, labels: Dict[str, str] = NOTHING, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, limit: int = 10000, count: int = 3, beam_width: Any = -1, limit_latest_versions: Any = -1, prescription: Optional[Prescription] = None, cli_parameters: Dict[str, Any] = NOTHING, stop_resolving: bool = False, log_iteration: int = 7500, 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_skip_package: Set[Tuple[str, str, str]] = NOTHING, log_step_not_acceptable: Set[Tuple[str, str, str]] = NOTHING, log_no_intersected: Set[Tuple[Tuple[str, str, str], str]] = NOTHING)[source]

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
property beam

Get beam for storing states.

beam_width
cli_parameters
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, labels: Optional[Dict[str, str]] = 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, prescription: Optional[Prescription] = None, cli_parameters: Optional[Dict[str, Any]] = None) → Resolver[source]

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, labels: Optional[Dict[str, str]] = 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, prescription: Optional[Prescription] = None, cli_parameters: Optional[Dict[str, Any]] = None) → Resolver[source]

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

graph
labels
library_usage
limit
limit_latest_versions
log_iteration
pipeline
plot() → matplotlib.figure.Figure[source]

Plot history captured during the resolution process.

predictor
prescription
project
recommendation_type
resolve(*, with_devel: bool = True, user_stack_scoring: bool = True) → thoth.adviser.report.Report[source]

Resolve software stacks and return resolver report.

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

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, *, with_devel: bool = True, user_stack_scoring: bool = True) → int[source]

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(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Sieve base class implementation.

static is_sieve_unit_type() → bool[source]

Check if this unit is of type sieve.

abstract 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.

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)[source]

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]]][source]

Fetch releases for the given package name.

graph
runtime_environment
class thoth.adviser.solver.PackageVersionDependencyParser[source]

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][source]

Parse the given list of PackageVersion objects.

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

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)[source]

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]][source]

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=None, *, advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None, advised_manifest_changes: List[List[Dict[str, Any]]] = NOTHING, justification: List[Dict[str, str]] = NOTHING)[source]

Bases: object

Implementation of an adviser’s state in state space.

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

Add new entries to the justification field.

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

Add a resolved dependency into the state.

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

Add unresolved dependency into the state.

advised_manifest_changes
advised_runtime_environment
clone() → thoth.adviser.state.State[source]

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[source]

Create an initial state out of direct dependencies.

get_first_unresolved_dependency(dependency_name: Optional[str] = None) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

get_random_first_unresolved_dependency(dependency_name: Optional[str] = None) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

get_random_unresolved_dependency(dependency_name: Optional[str] = None, prefer_recent: bool = True) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

is_final() → bool[source]

Check if the given state is a final state.

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

Iterate over resolved dependencies.

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

Iterate over unresolved dependencies.

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

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[source]

Remove the given unresolved dependency from state.

remove_unresolved_dependency_subtree(package_name: str) → None[source]

Remove the whole dependency sub-tree from the state.

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

Set unresolved dependencies - any unresolved dependencies will be overwritten.

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

Convert state to a dict representation.

unresolved_dependencies
update_unresolved_dependencies(dependencies: Dict[str, List[Tuple[str, str, str]]]) → None[source]

Update unresolved dependencies respecting the ones passed in as parameters.

thoth.adviser.step module

A base class for implementing steps.

class thoth.adviser.step.Step(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

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_DEFAULT = {'multi_package_resolution': False, 'package_name': None}
CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'multi_package_resolution': <class 'bool'>}, 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.

abstract 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.

thoth.adviser.stride module

A base class for implementing strides.

class thoth.adviser.stride.Stride(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Stride base class implementation.

static is_stride_unit_type() → bool[source]

Check if this unit is of type stride.

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

Run main entry-point for strides.

thoth.adviser.unit module

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

class thoth.adviser.unit.Unit(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: object

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

CONFIGURATION_DEFAULT = {'package_name': None}
CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None)}, extra=PREVENT_EXTRA, required=False) object>
classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[None, None, None][source]

Assign context to all units.

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]][source]

Get AICoE specific configuration encoded in the AICoE index URL.

classmethod get_unit_name() → str[source]

Get name of the unit.

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

Check if the given package-version is AICoE release.

static is_boot_unit_type() → bool[source]

Check if the unit is of type boot.

static is_pseudonym_unit_type() → bool[source]

Check if the unit is of type pseudonym.

static is_sieve_unit_type() → bool[source]

Check if the unit is of type sieve.

static is_step_unit_type() → bool[source]

Check if the unit is of type step.

static is_stride_unit_type() → bool[source]

Check if the unit is of type step.

static is_wrap_unit_type() → bool[source]

Check if the unit is of type wrap.

property name

Get name of this pipeline unit.

post_run() → None[source]

Called after the resolution is finished.

This method should not raise any exception.

post_run_report(report)[source]

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[source]

Called before running any pipeline unit with context already assigned.

This method should not raise any exception.

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.

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

Turn this pipeline unit into its dictionary representation.

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

Set configuration for a pipeline unit.

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

thoth.adviser.unit_types module

Types of units.

thoth.adviser.utils module

Various helpers and utility functions.

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

Log the given message once.

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

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(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Wrap base class implementation.

static is_wrap_unit_type() → bool[source]

Check if this unit is of type wrap.

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

Run 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, *, keep_history: Any = None, beam_history: List[Tuple[int, Optional[float]]] = NOTHING)[source]

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(state: thoth.adviser.state.State) → None[source]

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

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

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_last() → Optional[thoth.adviser.state.State][source]

Get state that was added in the previous resolution round.

get_random() → thoth.adviser.state.State[source]

Get a random state from beam.

iter_states() → List[thoth.adviser.state.State][source]

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][source]

Iterate over sorted states.

keep_history
max() → thoth.adviser.state.State[source]

Return the highest rated state as kept in the beam.

new_iteration() → None[source]

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[source]

Plot temperature history of adaptive simulated annealing.

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

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

If index is not provided, pop the largest item kept in the beam.

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

Remove the given state from beam.

property size

Get the current size of beam.

width
wipe() → None[source]

Remove all states from beam.

class thoth.adviser.Boot(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Boot base class implementation.

static is_boot_unit_type() → bool[source]

Check if this unit is of type boot.

abstract run() → None[source]

Run main entry-point for boot units.

class thoth.adviser.Context(*, project: thoth.python.project.Project, graph: thoth.storages.graph.postgres.GraphDatabase, labels: Dict[str, str], 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, prescription: Optional[Prescription] = None, cli_parameters: Dict[str, Any] = NOTHING, stack_info: 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)[source]

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
cli_parameters
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][source]

Get the given package version registered to the context.

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

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

graph
is_adviser() → bool[source]

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

is_dependency_monkey() → bool[source]

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

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

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][source]

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

iteration
labels
library_usage
limit
package_versions
prescription
project
recommendation_type
register_accepted_final_state(state: thoth.adviser.state.State) → None[source]

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[source]

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[source]

Register the given package version to the context.

sources
stack_info
class thoth.adviser.DecisionType[source]

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>)[source]

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
plot() → matplotlib.figure.Figure[source]

Plot info from Dependency Monkey run.

property predictor

Get predictor for the current dependency monkey configuration.

resolve(*, with_devel: bool = True, user_stack_scoring: bool = False) → thoth.adviser.dm_report.DependencyMonkeyReport[source]

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)[source]

Bases: object

Report produced by a Dependency Monkey run.

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

Add a new response to response listing.

skipped
to_dict() → Dict[str, Any][source]

Convert report to a dict representation suitable for serialization.

class thoth.adviser.Ecosystem[source]

Bases: thoth.adviser.enums._ExtendedEnum

Ecosystem known to recommendation engine.

PYTHON = 1
class thoth.adviser.PipelineBuilder[source]

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[source]

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, labels: Dict[str, str], library_usage: Optional[Dict[str, Any]], prescription: Optional[Prescription], cli_parameters: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig[source]

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, labels: Dict[str, str], library_usage: Optional[Dict[str, Any]], prescription: Optional[Prescription], cli_parameters: Dict[str, Any]) → thoth.adviser.pipeline_config.PipelineConfig[source]

Get dependency-monkey’s pipeline configuration.

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

Load pipeline configuration from a file or a string.

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

Bases: object

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

property boots

Get all boots.

property boots_dict

Get boots as a dictionary mapping.

call_post_run() → None[source]

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

call_post_run_report(report)[source]

Call post-run method when report is generated.

call_pre_run() → None[source]

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

iter_units()[source]

Iterate over units present in the configuration.

iter_units_reversed()[source]

Iterate over units present in the configuration in a reversed order.

property pseudonyms

Get all pseudonyms.

property pseudonyms_dict

Get pseudonyms as a dictionary mapping.

property sieves

Get all sieves.

property sieves_dict

Get sieves as a dictionary mapping.

property steps

Get all steps.

property steps_dict

Get steps as a dictionary mapping.

property strides

Get all strides.

property strides_dict

Get strides as a dictionary mapping.

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

Return this pipeline configuration in a dict representation.

property wraps

Get all wraps.

property wraps_dict

Get wraps as a dictionary mapping.

class thoth.adviser.Predictor(*, keep_history: Any = None)[source]

Bases: object

A base class for implementing a predictor for resolver.

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

Assign context to predictor.

property context

Get context in which the unit runs in.

finalize_state(state_id: int) → None[source]

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

keep_history
classmethod obtain_default_configuration(config_option_name: str) → Any[source]

Get the default value assigned to the given configuration option of a predictor instance.

plot() → matplotlib.figure.Figure[source]

Plot information about predictor.

post_run() → None[source]

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

post_run_report(report: thoth.adviser.report.Report) → None[source]

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[source]

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]][source]

Run 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[source]

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, advised_manifest_changes: List[List[Dict[str, Any]]] = NOTHING)[source]

Bases: object

A representation of an advised stack.

advised_manifest_changes
advised_runtime_environment
classmethod from_final_state(*, context: thoth.adviser.context.Context, state: thoth.adviser.state.State) → thoth.adviser.product.Product[source]

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

justification
project
score
to_dict() → Dict[str, Any][source]

Convert this instance into a dictionary.

class thoth.adviser.PythonRecommendationOutput[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

PIPENV = 1
REQUIREMENTS = 2
class thoth.adviser.RecommendationType[source]

Bases: thoth.adviser.enums._ExtendedEnum

Recommendation generation respecting constraints on package-version level observations.

LATEST = 3
PERFORMANCE = 4
SECURITY = 5
STABLE = 1
TESTING = 2
class thoth.adviser.Report(*, count: int, pipeline: thoth.adviser.pipeline_config.PipelineConfig, resolver_iterations: int = 0, accepted_final_states_count: int = 0, discarded_final_states_count: int = 0, stack_info: Optional[List[Dict[str, Any]]] = None, heapq: List[Tuple[Tuple[float, int], thoth.adviser.product.Product]] = NOTHING, heapq_counter: int = 0)[source]

Bases: object

A report stating output of an adviser run.

accepted_final_states_count
add_product(product: thoth.adviser.product.Product) → bool[source]

Add adviser pipeline product to report.

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

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

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

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

pipeline
product_count() → int[source]

Get number of products stored in the report.

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

Set stack information.

property stack_info

Retrieve stack info of this report.

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

Convert pipeline report to a dict representation.

will_add_product_from_state(score: float) → bool[source]

Check if a product with the given score would be added.

class thoth.adviser.Resolver(*, pipeline: thoth.adviser.pipeline_config.PipelineConfig, project: thoth.python.project.Project, library_usage: Any, graph: thoth.storages.graph.postgres.GraphDatabase, predictor: thoth.adviser.predictor.Predictor, labels: Dict[str, str] = NOTHING, recommendation_type: Optional[thoth.adviser.enums.RecommendationType] = None, decision_type: Optional[thoth.adviser.enums.DecisionType] = None, limit: int = 10000, count: int = 3, beam_width: Any = -1, limit_latest_versions: Any = -1, prescription: Optional[Prescription] = None, cli_parameters: Dict[str, Any] = NOTHING, stop_resolving: bool = False, log_iteration: int = 7500, 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_skip_package: Set[Tuple[str, str, str]] = NOTHING, log_step_not_acceptable: Set[Tuple[str, str, str]] = NOTHING, log_no_intersected: Set[Tuple[Tuple[str, str, str], str]] = NOTHING)[source]

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
property beam

Get beam for storing states.

beam_width
cli_parameters
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, labels: Optional[Dict[str, str]] = 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, prescription: Optional[Prescription] = None, cli_parameters: Optional[Dict[str, Any]] = None) → Resolver[source]

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, labels: Optional[Dict[str, str]] = 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, prescription: Optional[Prescription] = None, cli_parameters: Optional[Dict[str, Any]] = None) → Resolver[source]

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

graph
labels
library_usage
limit
limit_latest_versions
log_iteration
pipeline
plot() → matplotlib.figure.Figure[source]

Plot history captured during the resolution process.

predictor
prescription
project
recommendation_type
resolve(*, with_devel: bool = True, user_stack_scoring: bool = True) → thoth.adviser.report.Report[source]

Resolve software stacks and return resolver report.

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

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(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Sieve base class implementation.

static is_sieve_unit_type() → bool[source]

Check if this unit is of type sieve.

abstract 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.

class thoth.adviser.State(score: float = 0.0, iteration: int = 0, unresolved_dependencies=NOTHING, resolved_dependencies=NOTHING, parent=None, *, advised_runtime_environment: Optional[thoth.common.config.runtime_environment.RuntimeEnvironment] = None, advised_manifest_changes: List[List[Dict[str, Any]]] = NOTHING, justification: List[Dict[str, str]] = NOTHING)[source]

Bases: object

Implementation of an adviser’s state in state space.

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

Add new entries to the justification field.

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

Add a resolved dependency into the state.

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

Add unresolved dependency into the state.

advised_manifest_changes
advised_runtime_environment
clone() → thoth.adviser.state.State[source]

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[source]

Create an initial state out of direct dependencies.

get_first_unresolved_dependency(dependency_name: Optional[str] = None) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

get_random_first_unresolved_dependency(dependency_name: Optional[str] = None) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

get_random_unresolved_dependency(dependency_name: Optional[str] = None, prefer_recent: bool = True) → Tuple[str, str, str][source]

Get a very first unresolved dependency tuple.

is_final() → bool[source]

Check if the given state is a final state.

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

Iterate over resolved dependencies.

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

Iterate over unresolved dependencies.

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

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[source]

Remove the given unresolved dependency from state.

remove_unresolved_dependency_subtree(package_name: str) → None[source]

Remove the whole dependency sub-tree from the state.

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

Set unresolved dependencies - any unresolved dependencies will be overwritten.

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

Convert state to a dict representation.

unresolved_dependencies
update_unresolved_dependencies(dependencies: Dict[str, List[Tuple[str, str, str]]]) → None[source]

Update unresolved dependencies respecting the ones passed in as parameters.

class thoth.adviser.Step(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

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_DEFAULT = {'multi_package_resolution': False, 'package_name': None}
CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None), 'multi_package_resolution': <class 'bool'>}, 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.

abstract 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.

class thoth.adviser.Stride(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Stride base class implementation.

static is_stride_unit_type() → bool[source]

Check if this unit is of type stride.

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

Run main entry-point for strides.

class thoth.adviser.Unit(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: object

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

CONFIGURATION_DEFAULT = {'package_name': None}
CONFIGURATION_SCHEMA = <Schema({'package_name': Any(<class 'str'>, None, msg=None)}, extra=PREVENT_EXTRA, required=False) object>
classmethod assigned_context(context: thoth.adviser.context.Context) → Generator[None, None, None][source]

Assign context to all units.

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]][source]

Get AICoE specific configuration encoded in the AICoE index URL.

classmethod get_unit_name() → str[source]

Get name of the unit.

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

Check if the given package-version is AICoE release.

static is_boot_unit_type() → bool[source]

Check if the unit is of type boot.

static is_pseudonym_unit_type() → bool[source]

Check if the unit is of type pseudonym.

static is_sieve_unit_type() → bool[source]

Check if the unit is of type sieve.

static is_step_unit_type() → bool[source]

Check if the unit is of type step.

static is_stride_unit_type() → bool[source]

Check if the unit is of type step.

static is_wrap_unit_type() → bool[source]

Check if the unit is of type wrap.

property name

Get name of this pipeline unit.

post_run() → None[source]

Called after the resolution is finished.

This method should not raise any exception.

post_run_report(report)[source]

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[source]

Called before running any pipeline unit with context already assigned.

This method should not raise any exception.

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.

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

Turn this pipeline unit into its dictionary representation.

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

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(*, unit_run: bool = False, configuration: Dict[str, Any] = NOTHING)[source]

Bases: thoth.adviser.unit.Unit

Wrap base class implementation.

static is_wrap_unit_type() → bool[source]

Check if this unit is of type wrap.

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

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