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() → Generator[[str, None], None][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() → Generator[[tuple, None], None][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.

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() → Generator[[str, None], None][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() → Generator[[tuple, None], None][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.

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

Bases: thoth.storages.exceptions.ThothStorageException

Raised if unable to determine solver information out of solver name run.

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.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.inspection_schema module

Schema definition for inspection results.

thoth.storages.inspections module

Adapter for persisting Amun inspection results.

class thoth.storages.inspections.InspectionResultsStore(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 Amun inspection results.

RESULT_TYPE = 'inspection'
SCHEMA = <Schema({'specification': <Schema({'base': <class 'str'>, 'build': <class 'dict'>, 'files': <class 'list'>, 'packages': <class 'list'>, 'python_packages': <class 'list'>, 'identifier': <class 'str'>, 'python': <class 'dict'>, 'run': <class 'dict'>, 'script': <class 'str'>}, extra=PREVENT_EXTRA, required=False) object>, 'created': <class 'str'>, 'build_log': <class 'str'>, 'job_log': <Schema({'exit_code': <class 'int'>, 'hwinfo': <class 'dict'>, 'script_sha256': <class 'str'>, 'stderr': <class 'str'>, 'stdout': <class 'dict'>, 'usage': <class 'dict'>}, extra=PREVENT_EXTRA, required=False) object>, 'inspection_id': <class 'str'>, 'status': <Schema({'build': <class 'dict'>, 'job': <class 'dict'>}, extra=PREVENT_EXTRA, required=False) object>}, extra=PREVENT_EXTRA, required=False) object>
filter_document_ids(inspection_identifiers: List[str]) → Dict[str, List][source]

Filter inspection document ids list according to the inspection identifiers selected.

Parameters

inspection_identifiers – list of identifier/s to filter inspection ids

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

Get id under which the given document will be stored.

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.package_analyses module

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

class thoth.storages.package_analyses.PackageAnalysisResultsStore(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 package analyzes.

RESULT_TYPE = 'package-analysis'
SCHEMA = None

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>, 'duration': <class 'int'>, 'hostname': <class 'str'>, '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>, '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() → Generator[[str, None], None][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() → Generator[[tuple, None], None][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.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.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: Optional[List[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → tuple[source]

Sync adviser documents into graph.

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

Sync image analysis documents into graph.

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

Sync dependency monkey reports into graph database.

thoth.storages.sync.sync_documents(document_ids: Optional[List[str]] = None, *, amun_api_url: Optional[str] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, inspection_only_graph_sync: bool = False, inspection_only_ceph_sync: bool = False, is_local: bool = False) → Dict[str, 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: Optional[List[str]] = None, *, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, amun_api_url: str, only_ceph_sync: bool = False, only_graph_sync: bool = False, is_local: bool = False) → tuple[source]

Sync observations made on Amun into graph database.

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

Sync package analysis documents into graph.

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

Sync provenance check documents into graph.

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

Sync solver documents into graph.

Module contents

Storage and database adapters for Thoth.