thoth.common package

Submodules

thoth.common.exceptions module

Exceptions used within thoth-common package.

exception thoth.common.exceptions.ConfigurationError[source]

Bases: thoth.common.exceptions.ThothCommonException

Raised on miss-configuration issues.

exception thoth.common.exceptions.NotFoundException[source]

Bases: thoth.common.exceptions.ThothCommonException

Raised if the given resource cannot be found.

exception thoth.common.exceptions.ThothCommonException[source]

Bases: Exception

A base class for Thoth-common exception hierarchy.

exception thoth.common.exceptions.WorkflowError[source]

Bases: thoth.common.exceptions.ThothCommonException

Raised on workflow related issues.

thoth.common.helpers module

Various utilities to make your life easier.

thoth.common.helpers.cwd(target: str) → Generator[[str, None], None][source]

Manage cwd in a pushd/popd fashion.

thoth.common.helpers.datetime2datetime_str(dt: Optional[datetime.datetime] = None) → str[source]

Create a string representation of a datetime.

thoth.common.helpers.datetime_str2timestamp(datetime_string: str) → int[source]

Parse datetime string represented in ISO format and return timestamp.

thoth.common.helpers.datetime_str_from_timestamp(timestamp: int) → str[source]

Convert a timestamp to datetime string representation.

thoth.common.helpers.format_datetime(dt: datetime.datetime) → str[source]

Return datetime string in default format.

thoth.common.helpers.get_default_datetime_format() → str[source]

Return default datetime format string.

thoth.common.helpers.get_service_account_token() → str[source]

Get token from service account token file.

thoth.common.helpers.parse_datetime(datetime_string: str) → datetime.datetime[source]

Parse datetime string represented in ISO format.

thoth.common.helpers.timestamp2datetime(timestamp: int) → datetime.datetime[source]

Convert a timestamp to datetime respecting UTC.

thoth.common.helpers.to_camel_case(obj: T) → T[source]

Convert dictionary keys to camelCase.

thoth.common.helpers.to_snake_case(obj: T) → T[source]

Convert dictionary keys to snake_case.

thoth.common.json module

Manipulation with JSON format.

class thoth.common.json.SafeJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: json.encoder.JSONEncoder

Convert objects to JSON, safely.

default(o: Any) → Any[source]

Convert an object to JSON, safely.

thoth.common.logging module

Logging configuration for whole Thoth.

thoth.common.logging.before_send_handler(event, hint)[source]

Filter the errors caught before sending to Sentry.

This function ignores the exceptions passed in as a environment variable in a comma separated manner.

thoth.common.logging.init_logging(logging_configuration: Optional[Dict[str, str]] = None, logging_env_var_start: Optional[str] = None) → None[source]

Initialize Thoth’s logging - respects all namespaces.

This function allows you to control logging facilities in Thoth. Logging can be configured via env variables so that deployment can respect your configuration. The structure of environment variables is THOTH_LOG_(MODULE) and the value of env variable states verbosity level as in the logging module (DEBUG, INFO, WARNING, ERROR).

>>> import os
>>> os.environ['THOTH_LOG_SOLVER']
WARNING

You can also specify more closely which sub-module logging you are configuring - submodules are separated with double dash:

>>> os.environ['THOTH_LOG_SOLVER__PYTHON']
DEBUG

You can also use arguments explicitly that override configuration in env variables (a shorthand for standard logging functionality):

>>> init_logging({'thoth.solver': 'DEBUG'})

Optionally you can specify prefix of the logging environment variable determining logging configuration via env vars (defaults to THOTH_LOG_).

thoth.common.openshift module

Handling OpenShift and Kubernetes objects across project.

class thoth.common.openshift.OpenShift(*, frontend_namespace: Optional[str] = None, middletier_namespace: Optional[str] = None, backend_namespace: Optional[str] = None, infra_namespace: Optional[str] = None, amun_infra_namespace: Optional[str] = None, amun_inspection_namespace: Optional[str] = None, kubernetes_api_url: Optional[str] = None, kubernetes_verify_tls: bool = True, openshift_api_url: Optional[str] = None, use_argo: bool = False, token: Optional[str] = None, token_file: Optional[str] = None, cert_file: Optional[str] = None, environ: Optional[Dict[str, str]] = None)[source]

Bases: object

Interaction with OpenShift Master.

can_run_workload(template: Dict[str, Any], namespace: str) → bool[source]

Check if the given (job) can be run in the given namespace based on mem, cpu and pod restrictions.

create_config_map(configmap_name: str, namespace: str, labels: Dict[str, str], data: Dict[str, str]) → str[source]

Create a ConfigMap in the given namespace.

static generate_id(prefix: Optional[str] = None, identifier: Optional[str] = None) → str[source]

Generate an identifier.

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

Get template for an adviser run.

get_build(build_id: str, namespace: str) → Dict[str, Any][source]

Get a build in the given namespace.

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

Get template for build analyze run.

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

Get template for build dependencies run.

get_build_log(build_id: str, namespace: str) → str[source]

Get log of a build in the given namespace.

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

Get template for build report run.

get_buildconfig(buildconfig_id: str, namespace: str) → Dict[str, Any][source]

Get a buildconfig in the given namespace.

get_configmap(configmap_id: str, namespace: str) → Dict[str, Any][source]

Get the given configmap in a namespace, return object representing config map.

get_configmaps(namespace: str, label_selector: str) → Dict[str, Any][source]

Get all configmaps in a namespace and select them by label.

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

Get template for a dependency monkey.

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

Get graph sync template.

get_image_streams(namespace: str, label_selector: str) → Dict[str, Any][source]

Get all image streams in a namespace and select them by label.

get_job_log(job_id: str, namespace: str) → Optional[str][source]

Get log of a pod running inside a job.

get_job_logs(job_id: str, namespace: str) → Optional[str][source]

Get log of a pod running inside a job.

get_job_status(job_id: str, namespace: str) → Dict[str, Any][source]

Get status of a Job and Pods created by the Job.

Raises

NotFoundError if no Job of such name is found in the namespace

get_job_status_count(label_selector: str, namespace: str) → Dict[str, int][source]

Count the number of Jobs per status in a specific namespace.

get_job_status_report(job_id: str, namespace: str) → Dict[str, Optional[str]][source]

Get status report of a Job and Pods created by the Job.

get_jobs(label_selector: str, namespace: Optional[str] = None) → Dict[str, Any][source]

Get all Jobs, select them by the provided label.

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

Get template for a Kebechet job.

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

Get template for package-analyzer.

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

Get template for package-extract.

get_pod_log(pod_id: str, namespace: Optional[str] = None) → Optional[str][source]

Get log of a pod based on assigned pod ID.

get_pod_status(pod_id: str, namespace: str) → Dict[str, Any][source]

Get status entry for a pod - low level routine.

get_pod_status_report(pod_id: str, namespace: str) → Dict[str, Optional[str]][source]

Get pod state and convert it to a user-friendly response.

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

Get template for a provenance checker.

get_quota_status(namespace: str) → Dict[str, Dict[str, Any]][source]

Get quota status.

For now, there is retrieved a very first quota specification and its status from resource quota list, we gather information about memory, CPU usage and number of pods.

get_solver_names() → List[str][source]

Retrieve name of solvers available in installation.

get_solver_template(solver: str) → Dict[str, Any][source]

Retrieve a solver template.

get_workflow(name: str = None, label_selector: str = None, namespace: Optional[str] = None) → Dict[str, Any][source]

Get Workflow from a namespace, use one of name or label_selector to identify which one to get.

get_workflow_status(name: str = None, label_selector: str = None, namespace: Optional[str] = None) → Dict[str, Any][source]

Get a Workflow status, use one of name or label_selector to identify which one to get.

oc_process(namespace: str, template: Dict[str, Any]) → Dict[str, Any][source]

Process the given template in OpenShift.

static parse_cpu_spec(cpu_spec: Optional[str]) → Optional[float][source]

Parse the given CPU requirement as used by OpenShift/Kubernetes.

static parse_memory_spec(memory_spec: Optional[str]) → Optional[float][source]

Parse the given CPU requirement as used by OpenShift/Kubernetes.

run_adviser(application_stack: Dict[Any, Any], output: str, recommendation_type: str, *, count: Optional[int] = None, limit: Optional[int] = None, runtime_environment: Optional[Dict[Any, Any]] = None, library_usage: Optional[Dict[Any, Any]] = None, origin: Optional[str] = None, is_s2i: Optional[bool] = None, dev: bool = False, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, github_event_type: Optional[str] = None, github_check_run_id: Optional[int] = None, github_installation_id: Optional[int] = None, github_base_repo_url: Optional[str] = None, re_run_adviser_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run adviser on the provided user input.

run_build_analyze(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build analyze on the provided user input.

run_build_dependencies(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build dependencies on the provided user input.

run_build_report(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build report on the provided user input.

run_dependency_monkey(requirements: Union[str, Dict[str, Any]], context: Dict[str, Any], *, pipeline: Optional[Dict[str, Any]] = None, stack_output: Optional[str] = None, report_output: Optional[str] = None, runtime_environment: Optional[Dict[Any, Any]] = None, seed: Optional[int] = None, dry_run: bool = False, decision: Optional[str] = None, count: Optional[int] = None, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run Dependency Monkey on the provided user input.

run_graph_sync(document_id: str, namespace: str, graph_sync_type: str, force_sync: str, *, template: Optional[Dict[str, Any]] = None, job_id: Optional[str] = None) → str[source]

Run the given graph sync.

run_kebechet_run_results(url: str, service: str, analysis_id: str, *, verbose: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Create a kebechet run-results job.

run_kebechet_run_url(url: str, service: str, *, verbose: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Create a kebechet run-url job.

run_package_analyzer(package_name: str, package_version: str, index_url: str, *, output: str, debug: bool = False, dry_run: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run package-analyzer to gather digests of packages and files present inside packages.

run_package_extract(image: str, output: str, *, environment_type: str, is_external: bool = True, origin: Optional[str] = None, registry_user: Optional[str] = None, registry_password: Optional[str] = None, verify_tls: bool = True, debug: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run package-extract analyzer to extract information from the provided image.

run_provenance_checker(application_stack: Dict[Any, Any], output: str, *, origin: Optional[str] = None, whitelisted_sources: Optional[List[str]] = None, debug: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run provenance checks on the provided user input.

run_solver(packages: str, output: str, solver: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = True, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run solver or all solver to solve the given requirements.

schedule_adviser(application_stack: Dict[Any, Any], output: str, recommendation_type: str, *, count: Optional[int] = None, limit: Optional[int] = None, runtime_environment: Optional[Dict[Any, Any]] = None, library_usage: Optional[Dict[Any, Any]] = None, origin: Optional[str] = None, is_s2i: Optional[bool] = None, dev: bool = False, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, github_event_type: Optional[str] = None, github_check_run_id: Optional[int] = None, github_installation_id: Optional[int] = None, github_base_repo_url: Optional[str] = None, re_run_adviser_id: Optional[str] = None) → str[source]

Schedule an adviser run.

schedule_all_solvers(packages: str, output: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = False) → List[str][source]

Schedule all solvers for the given packages.

schedule_build_analyze(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build analyze run.

schedule_build_dependencies(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build analyze run.

schedule_build_report(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build report run.

schedule_dependency_monkey(requirements: Union[str, Dict[str, Any]], context: Dict[str, Any], *, pipeline: Optional[Dict[str, Any]] = None, stack_output: Optional[str] = None, report_output: Optional[str] = None, runtime_environment: Optional[Dict[Any, Any]] = None, seed: Optional[int] = None, dry_run: bool = False, decision: Optional[str] = None, count: Optional[int] = None, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None) → str[source]

Schedule a dependency monkey run.

schedule_graph_refresh(namespace: Optional[str] = None) → str[source]

Schedule graph refresh job in frontend namespace by default.

schedule_graph_sync_adviser(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an adviser document.

schedule_graph_sync_build_report(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an build log analysis document.

schedule_graph_sync_dependency_monkey(document_id: str, *, force_sync: bool = False, namespace: str) → str[source]

Schedule a sync of a dependency monkey result.

schedule_graph_sync_inspection(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of inspection.

schedule_graph_sync_package_analyzer(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an package-analysis document.

schedule_graph_sync_package_extract(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of package-extract.

schedule_graph_sync_provenance_checker(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of a provenance checker result.

schedule_graph_sync_solver(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of solver result.

schedule_inspection(dockerfile: str, specification: Dict[str, Any], target: str, parameters: Dict[str, Any]) → str[source]

Schedule an inspection run.

schedule_kebechet_run_results(url: str, service: str, analysis_id: str, *, verbose: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a kebechet run.

schedule_kebechet_run_url(url: str, service: str, *, verbose: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a kebechet run.

schedule_kebechet_workflow(webhook_payload: Dict[str, Any]) → str[source]

Schedule Kebechet Workflow for a Webhook from GitHub App..

schedule_package_analyzer(package_name: str, package_version: str, index_url: str, *, output: str, debug: bool = False, dry_run: bool = False, job_id: Optional[str] = None) → str[source]

Schedule the given job run, the scheduled job is handled by workload operator based resources available.

schedule_package_extract(image: str, output: str, *, environment_type: str, is_external: bool = True, origin: Optional[str] = None, registry_user: Optional[str] = None, registry_password: Optional[str] = None, verify_tls: bool = True, debug: bool = False, job_id: Optional[str] = None) → str[source]

Schedule the given job run, the scheduled job is handled by workload operator based resources available.

schedule_provenance_checker(application_stack: Dict[Any, Any], output: str, *, origin: Optional[str] = None, whitelisted_sources: Optional[List[str]] = None, debug: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a provenance checker run.

schedule_solver(packages: str, output: str, solver: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = True, job_id: Optional[str] = None) → str[source]

Schedule the given solver.

schedule_thamos_workflow(github_event_type: str, github_check_run_id: int, github_installation_id: int, github_base_repo_url: str, github_head_repo_url: str, origin: str, revision: str, host: str) → str[source]

Schedule Thamos Advise Workflow for Qeb-Hwt GitHub App..

static set_template_parameters(template: Dict[str, Any], **parameters: Any) → None[source]

Set parameters in the template - replace existing ones or append to parameter list if not exist.

>>> set_template_parameters(template, THOTH_LOG_ADVISER='DEBUG')
property token

Access service account token mounted to the pod.

property use_argo

Return whether Argo Workflows are currently enabled.

property workflow_manager

Return WorkflowManager instance.

This property lazily initializes the WorkflowManager.

thoth.common.workflows module

Workflow management for Thoth.

class thoth.common.workflows.Workflow(api_version: str, kind: str, metadata: argo.workflows.client.models.v1alpha1_metadata.V1alpha1Metadata, spec: argo.workflows.client.models.v1alpha1_workflow_spec.V1alpha1WorkflowSpec, status: Optional[argo.workflows.client.models.v1alpha1_workflow_status.V1alpha1WorkflowStatus] = None)[source]

Bases: argo.workflows.client.models.v1alpha1_workflow.V1alpha1Workflow

Argo Workflow instance.

This is a subclass of argo.workflows V1alpha1Workflow model which provides a convenient set of methods to make workflow management easier.

classmethod from_dict(wf: Dict[str, Any], validate: bool = True) → thoth.common.workflows.Workflow[source]

Create a Workflow from a dict.

classmethod from_file(fp: Union[str, pathlib.Path], validate: bool = True) → thoth.common.workflows.Workflow[source]

Create a Workflow from a file.

classmethod from_string(wf: str, validate: bool = True) → thoth.common.workflows.Workflow[source]

Create a Workflow from a YAML string.

classmethod from_url(url: str, validate: bool = True) → thoth.common.workflows.Workflow[source]

Create a Workflow from a remote file.

property id

Get Workflow ID.

property name

Get Workflow name.

property validated

Return whether this workflow has been validated.

class thoth.common.workflows.WorkflowManager(openshift: Optional[thoth.common.openshift.OpenShift] = None, openshift_config: Optional[Mapping[str, str]] = None)[source]

Bases: object

Argo Workflow manager.

get_workflow(namespace: str, name: str) → Dict[str, Any][source]

Get Workflow in namespace by name.

get_workflow_and_tasks_status(namespace: str, name: str) → Dict[str, Any][source]

Get workflow and tasks status from a specific namespace.

get_workflow_info(namespace: str, name: str) → Dict[str, Any][source]

Get Workflow in namespace by name.

get_workflow_template(namespace: str, label_selector: str, *, parameters: Dict[str, str])[source]

Get Workflow template.

get_workflows(namespace: str, *, label_selector: Optional[str] = None) → Dict[str, Any][source]

Get Workflows in namespace.

get_workflows_and_tasks_status(namespace: str, label_selector: Optional[str] = None) → Dict[str, Any][source]

Get workflows and tasks status from a specific namespace.

get_workflows_info(namespace: str, *, label_selector: Optional[str] = None) → Dict[str, Any][source]

Get workflows info from a specific namespace.

submit_adviser_workflow(template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None) → str[source]

Submit Adviser Workflow.

submit_inspection_workflow(template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None, use_hw_template: bool = False) → str[source]

Submit the Inspection Workflow.

submit_kebechet_workflow(template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None) → str[source]

Submit Kebechet Workflow.

submit_solver_workflow(template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None) → str[source]

Submit Solver Workflow.

submit_thamos_workflow(template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None) → str[source]

Submit Thamos Advise Workflow for Qeb-Hwt GitHub App.

submit_workflow(namespace: str, wf: Union[argo.workflows.client.models.v1alpha1_workflow.V1alpha1Workflow, Dict[str, Any]], *, parameters: Optional[Dict[str, str]] = None, validate: bool = True) → str[source]

Submit an Argo Workflow to a given namespace.

Returns

Workflow ID

submit_workflow_from_template(namespace: str, label_selector: str, *, template_parameters: Optional[Dict[str, str]] = None, workflow_parameters: Optional[Dict[str, Any]] = None, workflow_namespace: Optional[str] = None) → str[source]

Retrieve and Submit Workflow from an OpenShift template.

Parameters

  • namespace –

    namespace to lookup the template in

    If workflow_namespace is not provided, this namespace is also implicitly the namespace where the workflow is submitted

  • label_selector – selector for the template, i.e. ‘template=workflow-template’

  • template_parameters – parameters for the template

  • workflow_parameters – parameters for the workflow

  • workflow_namespace – namespace to submit the workflow to

Module contents

Shared code across Thoth analyzers.

thoth.common.cwd(target: str) → Generator[[str, None], None][source]

Manage cwd in a pushd/popd fashion.

thoth.common.datetime2datetime_str(dt: Optional[datetime.datetime] = None) → str[source]

Create a string representation of a datetime.

thoth.common.datetime_str2timestamp(datetime_string: str) → int[source]

Parse datetime string represented in ISO format and return timestamp.

thoth.common.datetime_str_from_timestamp(timestamp: int) → str[source]

Convert a timestamp to datetime string representation.

thoth.common.get_service_account_token() → str[source]

Get token from service account token file.

class thoth.common.HardwareInformation(cpu_family: int = None, cpu_model: int = None)[source]

Bases: thoth.common.config.base.ConfigEntryBase

Representation for hardware related information.

cpu_family
cpu_model
thoth.common.init_logging(logging_configuration: Optional[Dict[str, str]] = None, logging_env_var_start: Optional[str] = None) → None[source]

Initialize Thoth’s logging - respects all namespaces.

This function allows you to control logging facilities in Thoth. Logging can be configured via env variables so that deployment can respect your configuration. The structure of environment variables is THOTH_LOG_(MODULE) and the value of env variable states verbosity level as in the logging module (DEBUG, INFO, WARNING, ERROR).

>>> import os
>>> os.environ['THOTH_LOG_SOLVER']
WARNING

You can also specify more closely which sub-module logging you are configuring - submodules are separated with double dash:

>>> os.environ['THOTH_LOG_SOLVER__PYTHON']
DEBUG

You can also use arguments explicitly that override configuration in env variables (a shorthand for standard logging functionality):

>>> init_logging({'thoth.solver': 'DEBUG'})

Optionally you can specify prefix of the logging environment variable determining logging configuration via env vars (defaults to THOTH_LOG_).

class thoth.common.OpenShift(*, frontend_namespace: Optional[str] = None, middletier_namespace: Optional[str] = None, backend_namespace: Optional[str] = None, infra_namespace: Optional[str] = None, amun_infra_namespace: Optional[str] = None, amun_inspection_namespace: Optional[str] = None, kubernetes_api_url: Optional[str] = None, kubernetes_verify_tls: bool = True, openshift_api_url: Optional[str] = None, use_argo: bool = False, token: Optional[str] = None, token_file: Optional[str] = None, cert_file: Optional[str] = None, environ: Optional[Dict[str, str]] = None)[source]

Bases: object

Interaction with OpenShift Master.

can_run_workload(template: Dict[str, Any], namespace: str) → bool[source]

Check if the given (job) can be run in the given namespace based on mem, cpu and pod restrictions.

create_config_map(configmap_name: str, namespace: str, labels: Dict[str, str], data: Dict[str, str]) → str[source]

Create a ConfigMap in the given namespace.

static generate_id(prefix: Optional[str] = None, identifier: Optional[str] = None) → str[source]

Generate an identifier.

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

Get template for an adviser run.

get_build(build_id: str, namespace: str) → Dict[str, Any][source]

Get a build in the given namespace.

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

Get template for build analyze run.

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

Get template for build dependencies run.

get_build_log(build_id: str, namespace: str) → str[source]

Get log of a build in the given namespace.

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

Get template for build report run.

get_buildconfig(buildconfig_id: str, namespace: str) → Dict[str, Any][source]

Get a buildconfig in the given namespace.

get_configmap(configmap_id: str, namespace: str) → Dict[str, Any][source]

Get the given configmap in a namespace, return object representing config map.

get_configmaps(namespace: str, label_selector: str) → Dict[str, Any][source]

Get all configmaps in a namespace and select them by label.

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

Get template for a dependency monkey.

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

Get graph sync template.

get_image_streams(namespace: str, label_selector: str) → Dict[str, Any][source]

Get all image streams in a namespace and select them by label.

get_job_log(job_id: str, namespace: str) → Optional[str][source]

Get log of a pod running inside a job.

get_job_logs(job_id: str, namespace: str) → Optional[str][source]

Get log of a pod running inside a job.

get_job_status(job_id: str, namespace: str) → Dict[str, Any][source]

Get status of a Job and Pods created by the Job.

Raises

NotFoundError if no Job of such name is found in the namespace

get_job_status_count(label_selector: str, namespace: str) → Dict[str, int][source]

Count the number of Jobs per status in a specific namespace.

get_job_status_report(job_id: str, namespace: str) → Dict[str, Optional[str]][source]

Get status report of a Job and Pods created by the Job.

get_jobs(label_selector: str, namespace: Optional[str] = None) → Dict[str, Any][source]

Get all Jobs, select them by the provided label.

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

Get template for a Kebechet job.

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

Get template for package-analyzer.

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

Get template for package-extract.

get_pod_log(pod_id: str, namespace: Optional[str] = None) → Optional[str][source]

Get log of a pod based on assigned pod ID.

get_pod_status(pod_id: str, namespace: str) → Dict[str, Any][source]

Get status entry for a pod - low level routine.

get_pod_status_report(pod_id: str, namespace: str) → Dict[str, Optional[str]][source]

Get pod state and convert it to a user-friendly response.

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

Get template for a provenance checker.

get_quota_status(namespace: str) → Dict[str, Dict[str, Any]][source]

Get quota status.

For now, there is retrieved a very first quota specification and its status from resource quota list, we gather information about memory, CPU usage and number of pods.

get_solver_names() → List[str][source]

Retrieve name of solvers available in installation.

get_solver_template(solver: str) → Dict[str, Any][source]

Retrieve a solver template.

get_workflow(name: str = None, label_selector: str = None, namespace: Optional[str] = None) → Dict[str, Any][source]

Get Workflow from a namespace, use one of name or label_selector to identify which one to get.

get_workflow_status(name: str = None, label_selector: str = None, namespace: Optional[str] = None) → Dict[str, Any][source]

Get a Workflow status, use one of name or label_selector to identify which one to get.

oc_process(namespace: str, template: Dict[str, Any]) → Dict[str, Any][source]

Process the given template in OpenShift.

static parse_cpu_spec(cpu_spec: Optional[str]) → Optional[float][source]

Parse the given CPU requirement as used by OpenShift/Kubernetes.

static parse_memory_spec(memory_spec: Optional[str]) → Optional[float][source]

Parse the given CPU requirement as used by OpenShift/Kubernetes.

run_adviser(application_stack: Dict[Any, Any], output: str, recommendation_type: str, *, count: Optional[int] = None, limit: Optional[int] = None, runtime_environment: Optional[Dict[Any, Any]] = None, library_usage: Optional[Dict[Any, Any]] = None, origin: Optional[str] = None, is_s2i: Optional[bool] = None, dev: bool = False, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, github_event_type: Optional[str] = None, github_check_run_id: Optional[int] = None, github_installation_id: Optional[int] = None, github_base_repo_url: Optional[str] = None, re_run_adviser_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run adviser on the provided user input.

run_build_analyze(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build analyze on the provided user input.

run_build_dependencies(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build dependencies on the provided user input.

run_build_report(document_id: str, output: str, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run build report on the provided user input.

run_dependency_monkey(requirements: Union[str, Dict[str, Any]], context: Dict[str, Any], *, pipeline: Optional[Dict[str, Any]] = None, stack_output: Optional[str] = None, report_output: Optional[str] = None, runtime_environment: Optional[Dict[Any, Any]] = None, seed: Optional[int] = None, dry_run: bool = False, decision: Optional[str] = None, count: Optional[int] = None, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run Dependency Monkey on the provided user input.

run_graph_sync(document_id: str, namespace: str, graph_sync_type: str, force_sync: str, *, template: Optional[Dict[str, Any]] = None, job_id: Optional[str] = None) → str[source]

Run the given graph sync.

run_kebechet_run_results(url: str, service: str, analysis_id: str, *, verbose: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Create a kebechet run-results job.

run_kebechet_run_url(url: str, service: str, *, verbose: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Create a kebechet run-url job.

run_package_analyzer(package_name: str, package_version: str, index_url: str, *, output: str, debug: bool = False, dry_run: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run package-analyzer to gather digests of packages and files present inside packages.

run_package_extract(image: str, output: str, *, environment_type: str, is_external: bool = True, origin: Optional[str] = None, registry_user: Optional[str] = None, registry_password: Optional[str] = None, verify_tls: bool = True, debug: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run package-extract analyzer to extract information from the provided image.

run_provenance_checker(application_stack: Dict[Any, Any], output: str, *, origin: Optional[str] = None, whitelisted_sources: Optional[List[str]] = None, debug: bool = False, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run provenance checks on the provided user input.

run_solver(packages: str, output: str, solver: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = True, job_id: Optional[str] = None, template: Optional[Dict[str, Any]] = None) → str[source]

Run solver or all solver to solve the given requirements.

schedule_adviser(application_stack: Dict[Any, Any], output: str, recommendation_type: str, *, count: Optional[int] = None, limit: Optional[int] = None, runtime_environment: Optional[Dict[Any, Any]] = None, library_usage: Optional[Dict[Any, Any]] = None, origin: Optional[str] = None, is_s2i: Optional[bool] = None, dev: bool = False, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None, github_event_type: Optional[str] = None, github_check_run_id: Optional[int] = None, github_installation_id: Optional[int] = None, github_base_repo_url: Optional[str] = None, re_run_adviser_id: Optional[str] = None) → str[source]

Schedule an adviser run.

schedule_all_solvers(packages: str, output: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = False) → List[str][source]

Schedule all solvers for the given packages.

schedule_build_analyze(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build analyze run.

schedule_build_dependencies(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build analyze run.

schedule_build_report(document_id: str, output: str, job_id: Optional[str] = None) → str[source]

Schedule an build report run.

schedule_dependency_monkey(requirements: Union[str, Dict[str, Any]], context: Dict[str, Any], *, pipeline: Optional[Dict[str, Any]] = None, stack_output: Optional[str] = None, report_output: Optional[str] = None, runtime_environment: Optional[Dict[Any, Any]] = None, seed: Optional[int] = None, dry_run: bool = False, decision: Optional[str] = None, count: Optional[int] = None, debug: bool = False, job_id: Optional[str] = None, limit_latest_versions: Optional[int] = None) → str[source]

Schedule a dependency monkey run.

schedule_graph_refresh(namespace: Optional[str] = None) → str[source]

Schedule graph refresh job in frontend namespace by default.

schedule_graph_sync_adviser(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an adviser document.

schedule_graph_sync_build_report(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an build log analysis document.

schedule_graph_sync_dependency_monkey(document_id: str, *, force_sync: bool = False, namespace: str) → str[source]

Schedule a sync of a dependency monkey result.

schedule_graph_sync_inspection(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of inspection.

schedule_graph_sync_package_analyzer(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of an package-analysis document.

schedule_graph_sync_package_extract(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of package-extract.

schedule_graph_sync_provenance_checker(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of a provenance checker result.

schedule_graph_sync_solver(document_id: str, *, force_sync: bool = False, namespace: Optional[str] = None) → str[source]

Schedule a sync of solver result.

schedule_inspection(dockerfile: str, specification: Dict[str, Any], target: str, parameters: Dict[str, Any]) → str[source]

Schedule an inspection run.

schedule_kebechet_run_results(url: str, service: str, analysis_id: str, *, verbose: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a kebechet run.

schedule_kebechet_run_url(url: str, service: str, *, verbose: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a kebechet run.

schedule_kebechet_workflow(webhook_payload: Dict[str, Any]) → str[source]

Schedule Kebechet Workflow for a Webhook from GitHub App..

schedule_package_analyzer(package_name: str, package_version: str, index_url: str, *, output: str, debug: bool = False, dry_run: bool = False, job_id: Optional[str] = None) → str[source]

Schedule the given job run, the scheduled job is handled by workload operator based resources available.

schedule_package_extract(image: str, output: str, *, environment_type: str, is_external: bool = True, origin: Optional[str] = None, registry_user: Optional[str] = None, registry_password: Optional[str] = None, verify_tls: bool = True, debug: bool = False, job_id: Optional[str] = None) → str[source]

Schedule the given job run, the scheduled job is handled by workload operator based resources available.

schedule_provenance_checker(application_stack: Dict[Any, Any], output: str, *, origin: Optional[str] = None, whitelisted_sources: Optional[List[str]] = None, debug: bool = False, job_id: Optional[str] = None) → str[source]

Schedule a provenance checker run.

schedule_solver(packages: str, output: str, solver: str, *, indexes: Optional[List[str]] = None, debug: bool = False, transitive: bool = True, job_id: Optional[str] = None) → str[source]

Schedule the given solver.

schedule_thamos_workflow(github_event_type: str, github_check_run_id: int, github_installation_id: int, github_base_repo_url: str, github_head_repo_url: str, origin: str, revision: str, host: str) → str[source]

Schedule Thamos Advise Workflow for Qeb-Hwt GitHub App..

static set_template_parameters(template: Dict[str, Any], **parameters: Any) → None[source]

Set parameters in the template - replace existing ones or append to parameter list if not exist.

>>> set_template_parameters(template, THOTH_LOG_ADVISER='DEBUG')
property token

Access service account token mounted to the pod.

property use_argo

Return whether Argo Workflows are currently enabled.

property workflow_manager

Return WorkflowManager instance.

This property lazily initializes the WorkflowManager.

class thoth.common.OperatingSystem(name: str = None, version: str = None)[source]

Bases: thoth.common.config.base.ConfigEntryBase

Representation for hardware related information.

name
version
thoth.common.parse_datetime(datetime_string: str) → datetime.datetime[source]

Parse datetime string represented in ISO format.

class thoth.common.RuntimeEnvironment(hardware: thoth.common.config.hardware_information.HardwareInformation, operating_system: thoth.common.config.operating_system.OperatingSystem, python_version: str = None, cuda_version: str = None, name: str = None, platform: str = None)[source]

Bases: object

An entry collapsing configuration options in the user configuration file.

cuda_version
classmethod from_dict(dict_: Optional[Dict[Any, Any]] = None) → thoth.common.config.runtime_environment.RuntimeEnvironment[source]

Parse one configuration entry from a dictionary.

hardware
is_fully_specified() → bool[source]

Check if the given runtime environment is fully specified.

classmethod load(content: Optional[str] = None) → thoth.common.config.runtime_environment.RuntimeEnvironment[source]

Load runtime environment information from file or from a JSON representation, transparently.

name
operating_system
platform
python_version
to_dict(without_none: bool = False) → Dict[str, Any][source]

Convert runtime environment configuration to a dict representation.

to_string() → str[source]

Convert runtime environment configuration to a string representation.

class thoth.common.SafeJSONEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: json.encoder.JSONEncoder

Convert objects to JSON, safely.

default(o: Any) → Any[source]

Convert an object to JSON, safely.