thoth.storages package

Submodules

thoth.storages.advisers module

Adapter for storing advices onto a persistence remote store.

class thoth.storages.advisers.AdvisersResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting recommendation results.

RESULT_TYPE = 'adviser'

thoth.storages.advisers_cache module

Adapter for retrieving and storing hashes for cached adviser results.

class thoth.storages.advisers_cache.AdvisersCacheStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.ceph_cache.CephCache

Adapter for retrieving and storing hashes for cached adviser results.

RESULT_TYPE = 'adviser-cache'

thoth.storages.analyses module

Adapter for storing analysis results onto a persistence remote store.

class thoth.storages.analyses.AnalysisResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Store results of image analyzes.

RESULT_TYPE = 'analysis'

thoth.storages.analyses_by_digest module

Adapter for persisting Amun inspection results.

class thoth.storages.analyses_by_digest.AnalysisByDigest(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for storing image analysis results by its digest.

RESULT_TYPE = 'analysis-by-digest'
SCHEMA = None
store_document(image_digest: str, analysis_info: str) → dict[source]

Store the given document representing analysis info onto Ceph.

thoth.storages.analyses_cache module

Adapter for retrieving and storing hashes for fast image lookups.

class thoth.storages.analyses_cache.AnalysesCacheStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.ceph_cache.CephCache

Adapter for retrieving and storing hashes for fast image lookups.

This adapter is used to store simple files on Ceph so that there can be quickly checked whether the given image was analyzed based in its SHA and which analysis corresponds to the analyzed image.

RESULT_TYPE = 'analysis-cache'

thoth.storages.base module

A base class for implementing storage adapters.

class thoth.storages.base.StorageBase[source]

Bases: object

A base class for implementing storage adapters.

connect() → None[source]

Connect the given storage adapter.

is_connected() → bool[source]

Check if the given database adapter is in connected state.

thoth.storages.buildlogs module

Adapter for storing build logs.

class thoth.storages.buildlogs.BuildLogsStore(deployment_name=None, *, bucket_prefix: str = None, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None)[source]

Bases: thoth.storages.base.StorageBase

Adapter for storing build logs.

RESULT_TYPE = 'buildlogs'
connect() → None[source]

Connect the given storage adapter.

get_document_listing() → typing.Generator[[str, NoneType], NoneType][source]

Get listing of documents stored on the Ceph.

is_connected() → bool[source]

Check if the given database adapter is in connected state.

iterate_results() → typing.Generator[[tuple, NoneType], NoneType][source]

Iterate over results available in the Ceph.

retrieve_document(document_id: str) → dict[source]

Retrieve a document from Ceph by its id.

store_document(document: dict) → str[source]

Store the given document in Ceph.

thoth.storages.buildlogs_analyses module

Adapter for storing buildlog analysis results onto a persistence remote store.

class thoth.storages.buildlogs_analyses.BuildLogsAnalysisResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Store results of buildlog analyzes.

RESULT_TYPE = 'buildlogs-analysis'

thoth.storages.buildlogs_analyses_cache module

Adapter for retrieving and storing hashes for fast buildlog lookups.

class thoth.storages.buildlogs_analyses_cache.BuildLogsAnalysesCacheStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.ceph_cache.CephCache

Adapter for retrieving and storing hashes for fast buildlog lookups.

This adapter is used to store simple files on Ceph so that there can be quickly checked whether the given buildlog was analyzed based in its SHA and which analysis corresponds to the analyzed buildlogs.

RESULT_TYPE = 'buildlogs-analysis-cache'

thoth.storages.ceph module

Adapter for Ceph distributed object storage.

class thoth.storages.ceph.CephStore(prefix, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None)[source]

Bases: thoth.storages.base.StorageBase

Adapter for storing and retrieving data from Ceph - low level API.

check_connection() → None[source]

Ceph Connection Check.

Check whether the given connection to the Ceph is alive and healthy, raise an exception if not.

connect() → None[source]

Create a connection to the remote Ceph.

delete(object_key: str) → None[source]

Delete the given object from Ceph.

static dict2blob(dictionary: dict) → bytes[source]

Encode a dictionary to a blob so it can be stored on Ceph.

document_exists(document_id: str) → bool[source]

Check if the there is an object with the given key in bucket.

This check does only HEAD request.

get_document_listing() → typing.Generator[[str, NoneType], NoneType][source]

Get listing of documents stored on the Ceph.

is_connected() → bool[source]

Check whether adapter is connected to the remote Ceph storage.

iterate_results() → typing.Generator[[tuple, NoneType], NoneType][source]

Iterate over results available in the Ceph.

retrieve_blob(object_key: str) → bytes[source]

Retrieve remote object content.

retrieve_document(document_id: str) → dict[source]

Retrieve a dictionary stored as JSON from S3.

store_blob(blob: bytes, object_key: str) → dict[source]

Store a blob on Ceph.

store_document(document: dict, document_id: str) → dict[source]

Store a document (dict) onto Ceph.

store_file(document_path: str, document_id: str) → dict[source]

Store a file on Ceph.

thoth.storages.ceph_cache module

A base class for implementing caches based on Ceph.

class thoth.storages.ceph_cache.CephCache(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

A base class implementing cache interface.

retrieve_document_record(document_id: str) → dict[source]

Check whether the given record exists in the cache for the requested document.

store_document_record(document_id: str, document: dict) → str[source]

Store the given document record in the cache.

thoth.storages.cli module

A CLI client to thoth-storages library.

thoth.storages.dependency_monkey_reports module

Adapter for persisting reports from Dependency Monkey runs.

class thoth.storages.dependency_monkey_reports.DependencyMonkeyReportsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting reports from Dependency Monkey runs.

RESULT_TYPE = 'dependency-monkey-reports'
iterate_inspection_ids() → str[source]

Iterate over all inspection ids that were run.

thoth.storages.exceptions module

Exceptions for storage adapters and storage handling.

exception thoth.storages.exceptions.AlreadyConnected[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if trying to connect on already connected adapter.

exception thoth.storages.exceptions.CacheMiss[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if the requested document was not found in the cache.

exception thoth.storages.exceptions.CudaVersionDoesNotMatch[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if the cuda versions from txt file and nvcc command is different.

exception thoth.storages.exceptions.DatabaseNotInitialized[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if trying to perform operations on un-initialized database schema.

exception thoth.storages.exceptions.DistutilsKeyNotKnown[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if a distutils in Python Package metadata is not known.

exception thoth.storages.exceptions.MultipleFoundError[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if there are multiple entities in the graph database when querying for a single one.

exception thoth.storages.exceptions.NoDocumentIdError[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if document id is not found in the document.

exception thoth.storages.exceptions.NotConnected[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if there was no connection established when communicating with a storage.

exception thoth.storages.exceptions.NotFoundError[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if the given artifact cannot be found.

exception thoth.storages.exceptions.PerformanceIndicatorNotRegistered[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if a performance indicator model which is about to be synced was not found.

exception thoth.storages.exceptions.PythonIndexNotProvided[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if an attempt to insert a package without an index.

exception thoth.storages.exceptions.PythonIndexNotRegistered[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if an attempt to insert a Python package with an index not being registered to the system.

exception thoth.storages.exceptions.PythonPackageMetadataAttributeMissing[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if a Python Package metadata attribute is not included in the class attributes.

exception thoth.storages.exceptions.SchemaError[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if trying to store document with invalid schema.

exception thoth.storages.exceptions.SolverNotRun[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if an attempt to insert a package which was not solved.

exception thoth.storages.exceptions.SortTypeQueryError[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if a sort key used in a query is not known.

exception thoth.storages.exceptions.ThothStorageException[source]

Bases: Exception

A base exception for Thoth storage exception hierarchy.

exception thoth.storages.exceptions.UnboundModel[source]

Bases: thoth.storages.exceptions.ThothStorageException

Raised if there are done operations on models which do not link to objects stored inside database.

thoth.storages.graph_backup module

Adapter for storing graph database backups.

class thoth.storages.graph_backup.GraphBackupStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for storing graph database backups.

GRAPH_BACKUP_STORE_ROTATE = 21
RESULT_TYPE = 'graph-backup'
store_dump(dump_file_path: str) → str[source]

Store the given dump, maintain a fixed set of dumps .

thoth.storages.inspection_schema module

Schema definition for inspection results.

thoth.storages.inspections module

Adapter for persisting Amun inspection results.

class thoth.storages.inspections.InspectionBuildsStore(inspection_id: str) → None[source]

Bases: thoth.storages.inspections._InspectionBase

An adapter for manipulating with inspection builds.

retrieve_dockerfile() → str[source]

Retrieve Dockerfile used during the build.

retrieve_log() → str[source]

Retrieve logs (stdout together with stderr) reported during the build.

retrieve_specification() → typing.Dict[str, typing.Any][source]

Retrieve specification used for the build, captures also run specification.

class thoth.storages.inspections.InspectionResultsStore(inspection_id: str) → None[source]

Bases: thoth.storages.inspections._InspectionBase

An adapter for manipulating with inspection results.

classmethod get_document_id(document: typing.Dict[str, typing.Any]) → str[source]

Get id under which the given document will be stored.

get_results_count() → int[source]

Obtain number of results produced during inspection run.

iter_inspection_results() → typing.Generator[[typing.Dict[str, typing.Any], NoneType], NoneType][source]

Iterate over inspection results.

retrieve_hwinfo(item: int) → typing.Dict[str, typing.Any][source]

Obtain hardware information for the given inspection run.

retrieve_log(item: int) → str[source]

Obtain log for the given inspection run.

retrieve_result(item: int) → typing.Dict[str, typing.Any][source]

Obtain the actual result for the given inspection run.

class thoth.storages.inspections.InspectionStore(inspection_id: str) → None[source]

Bases: object

Adapter for manipulating with Amun inspections.

build
check_connection()[source]

Check connections of this adapter.

connect() → None[source]

Connect this adapter.

exists() → bool[source]

Check if the given inspection exists.

classmethod get_inspection_count() → int[source]

Get number of inspection stored.

inspection_id
is_connected() → bool[source]

Check if this adapter is connected.

classmethod iter_inspections() → typing.Generator[[str, NoneType], NoneType][source]

Iterate over inspection ids stored.

results
retrieve_specification() → typing.Dict[str, typing.Any][source]

Retrieve specification used for this inspection.

thoth.storages.observations module

Adapter for storing and retrieving observation reports.

class thoth.storages.observations.ObservationResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Store resulting observation reports.

RESULT_TYPE = 'observations'

thoth.storages.provenance module

Adapter for storing and retrieving provenance reports.

class thoth.storages.provenance.ProvenanceResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Store results of provenance reports.

RESULT_TYPE = 'provenance'

thoth.storages.provenance_cache module

Adapter for retrieving and storing hashes for cached provenance checker results.

class thoth.storages.provenance_cache.ProvenanceCacheStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.ceph_cache.CephCache

Adapter for retrieving and storing hashes for cached provenance checker results.

RESULT_TYPE = 'provenance-cache'

thoth.storages.result_base module

Adapter for storing analysis results onto a persistence remote store.

class thoth.storages.result_base.ResultStorageBase(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.base.StorageBase

Adapter base for storing results.

RESULT_TYPE = ''
SCHEMA = <Schema({'metadata': <Schema({'analyzer': <class 'str'>, 'analyzer_version': <class 'str'>, 'arguments': <class 'dict'>, 'datetime': <thoth.storages.result_schema.Datetime object>, 'distribution': <Schema({'codename': <class 'str'>, 'id': <class 'str'>, 'like': <class 'str'>, 'version': <class 'str'>, 'version_parts': {'build_number': <class 'str'>, 'major': <class 'str'>, 'minor': <class 'str'>}}, extra=PREVENT_EXTRA, required=False) object>, 'document_id': <class 'str'>, 'duration': <class 'int'>, 'hostname': <class 'str'>, 'os_release': <Schema({'redhat_bugzilla_product': <class 'str'>, 'redhat_bugzilla_product_version': <class 'str'>, 'redhat_support_product': <class 'str'>, 'redhat_support_product_version': <class 'str'>, 'variant_id': <class 'str'>, 'id': <class 'str'>, 'name': <class 'str'>, 'platform_id': <class 'str'>, 'version_id': <class 'str'>, 'version': <class 'str'>}, extra=PREVENT_EXTRA, required=False) object>, 'python': <Schema({'api_version': <class 'int'>, 'implementation_name': <class 'str'>, 'major': <class 'int'>, 'micro': <class 'int'>, 'minor': <class 'int'>, 'releaselevel': <class 'str'>, 'serial': <class 'int'>}, extra=PREVENT_EXTRA, required=False) object>, 'thoth_deployment_name': <class 'str'>, 'timestamp': <class 'int'>}, extra=PREVENT_EXTRA, required=False) object>, 'result': <class 'object'>}, extra=PREVENT_EXTRA, required=False) object>
connect() → None[source]

Connect the given storage adapter.

document_exists(document_id: str) → bool[source]

Check if the there is an object with the given key in bucket.

get_document_count() → int[source]

Get number of documents present.

classmethod get_document_id(document: dict) → str[source]

Get document id under which the given document should be stored.

get_document_listing() → typing.Generator[[str, NoneType], NoneType][source]

Get listing of documents available in Ceph as a generator.

is_connected() → bool[source]

Check if the given database adapter is in connected state.

iterate_results() → typing.Generator[[tuple, NoneType], NoneType][source]

Iterate over results available in the Ceph.

retrieve_document(document_id: str) → dict[source]

Retrieve a document from Ceph by its id.

store_document(document: dict) → str[source]

Store the given document in Ceph.

store_file(file_path: str, file_id: str) → str[source]

Store the given file in Ceph.

thoth.storages.result_schema module

Schema definition for analyzer results.

class thoth.storages.result_schema.Datetime[source]

Bases: object

Check datetime fields against ISO format.

thoth.storages.revsolvers module

Adapter for storing reverse solver results onto a persistence remote store.

class thoth.storages.revsolvers.RevSolverResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting reverse solver results.

RESULT_TYPE = 'revsolver'

thoth.storages.security_indicators module

Adapter for persisting Security Indicator results.

class thoth.storages.security_indicators.SIAggregatedStore(security_indicator_id: str) → None[source]

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating with security-indicators aggregated.

document_exists() → bool[source]

Check if the there is an object with the given key in bucket.

retrieve_document() → typing.Dict[str, typing.Any][source]

Retrieve SI aggregated document.

class thoth.storages.security_indicators.SIBanditStore(security_indicator_id: str) → None[source]

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating with security-indicators bandit.

document_exists() → bool[source]

Check if the there is an object with the given key in bucket.

retrieve_document() → typing.Dict[str, typing.Any][source]

Retrieve SI bandit document.

class thoth.storages.security_indicators.SIClocStore(security_indicator_id: str) → None[source]

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating with security-indicators cloc.

document_exists() → bool[source]

Check if the there is an object with the given key in bucket.

retrieve_document() → typing.Dict[str, typing.Any][source]

Retrieve SI cloc document.

class thoth.storages.security_indicators.SecurityIndicatorsResultsStore(security_indicator_id: str) → None[source]

Bases: object

Adapter for manipulating with Security Indicators.

aggregated
bandit
check_connection()[source]

Check connections of this adapter.

cloc
connect() → None[source]

Connect this adapter.

classmethod get_security_indicators_count() → int[source]

Get number of security_indicators stored.

is_connected() → bool[source]

Check if this adapter is connected.

classmethod iter_security_indicators() → typing.Generator[[str, NoneType], NoneType][source]

Iterate over security_indicators ids stored.

security_indicator_id

thoth.storages.solvers module

Adapter for storing solver results onto a persistence remote store.

class thoth.storages.solvers.SolverResultsStore(deployment_name=None, *, host: str = None, key_id: str = None, secret_key: str = None, bucket: str = None, region: str = None, prefix: str = None)[source]

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting solver results.

RESULT_TYPE = 'solver'
static get_solver_name_from_document_id(solver_document_id: str) → str[source]

Retrieve solver name from solver’s document id.

thoth.storages.sync module

Routines for syncing data from Ceph into graph database.

thoth.storages.sync.sync_adviser_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync adviser documents into graph.

thoth.storages.sync.sync_analysis_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync image analysis documents into graph.

thoth.storages.sync.sync_build_log_analysis_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync build log analysis documents into graph.

thoth.storages.sync.sync_dependency_monkey_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync dependency monkey reports into graph database.

thoth.storages.sync.sync_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, *, amun_api_url: typing.Union[str, NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, inspection_only_graph_sync: bool = False, inspection_only_ceph_sync: bool = False, is_local: bool = False) → typing.Dict[str, typing.Tuple[[int, int, int], int]][source]

Sync documents based on document type.

If no list of document ids is provided, all documents will be synced. >>> from thoth.storages.sync import sync_documents >>> sync_documents([“adviser-efa7213babd12911”, “package-extract-f8e354d9597a1203”])

thoth.storages.sync.sync_inspection_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, *, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync observations made on Amun into graph database.

thoth.storages.sync.sync_provenance_checker_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync provenance check documents into graph.

thoth.storages.sync.sync_revsolver_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync reverse solver documents into graph.

thoth.storages.sync.sync_security_indicators_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync security indicators results into graph.

thoth.storages.sync.sync_solver_documents(document_ids: typing.Union[typing.List[str], NoneType] = None, force: bool = False, graceful: bool = False, graph: typing.Union[thoth.storages.graph.postgres.GraphDatabase, NoneType] = None, is_local: bool = False) → tuple[source]

Sync solver documents into graph.

Module contents

Storage and database adapters for Thoth.