thoth.storages package¶

Subpackages¶

thoth.storages.graph package

Submodules¶

thoth.storages.advisers module¶

Adapter for storing advices onto a persistence remote store.

class thoth.storages.advisers.AdvisersResultsStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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¶

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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[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_cache module¶

Adapter for retrieving and storing hashes for fast buildlog lookups.

class thoth.storages.buildlogs_analyses_cache.BuildLogsAnalysesCacheStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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.buildlogs_parsed module¶

Adapter for manipulating with parsed buildlogs.

class thoth.storages.buildlogs_parsed.BuildLogsParsedResultsStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[str] = None)[source]¶

Bases: thoth.storages.result_base.ResultStorageBase

Store results of buildlog parser.

RESULT_TYPE = 'buildlogs-parsed'¶

thoth.storages.ceph module¶

Adapter for Ceph distributed object storage.

class thoth.storages.ceph.CephStore(prefix, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[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(prefix_addition: Optional[str] = '') → 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(prefix_addition: Optional[str] = None) → 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.

retrieve_document_attr(object_key: str, attr: str) → Any[source]¶: Retrieve the given attribute of a document 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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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.

retrieve_document_ttl(document_id: str) → float[source]¶: Retrieve the TTL for a cached document.

store_document_record(document_id: str, document: dict) → None[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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() → Iterable[str][source]¶: Iterate over all inspection ids that were run.

thoth.storages.dependency_monkey_requests module¶

Adapter for persisting requests to Dependency Monkey runs.

class thoth.storages.dependency_monkey_requests.DependencyMonkeyRequestsStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[str] = None)[source]¶

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting requests to Dependency Monkey runs.

RESULT_TYPE = 'dependency-monkey-requests'¶

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 when a method used requires just one present.

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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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.inspections module¶

Adapter for persisting Amun inspection results.

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

Bases: thoth.storages.inspections._InspectionBase

An adapter for retrieving inspection builds.

ceph¶

inspection_id¶

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() → Dict[str, Any][source]¶: Retrieve specification used for the build, captures also run specification.

sub_prefix = 'build'¶

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

Bases: thoth.storages.inspections._InspectionBase

An adapter for manipulating with inspection results.

ceph¶

classmethod get_document_id(document: Dict[str, 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.

inspection_id¶

iter_inspection_results() → Generator[Dict[str, Any], None, None][source]¶: Iterate over inspection results.

retrieve_hwinfo(item: int) → Dict[str, 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) → Dict[str, Any][source]¶: Obtain the actual result for the given inspection run.

sub_prefix = 'results'¶

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

Bases: object

Adapter for manipulating 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() → Generator[str, None, None][source]¶: Iterate over inspection ids stored.

results¶

retrieve_specification() → Dict[str, Any][source]¶: Retrieve specification used for this inspection.

thoth.storages.logs module¶

Adapter for accessing Argo Workflow logs.

class thoth.storages.logs.WorkflowLogsStore(deployment_name: Optional[str] = None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[str] = None)[source]¶

Bases: thoth.storages.base.StorageBase

Access logs stored by Argo Workflows.

connect() → None[source]¶: Connect to Ceph.

get_log(workflow_id: str) → str[source]¶: Obtain log from the given workflow.

thoth.storages.observations module¶

Adapter for storing and retrieving observation reports.

class thoth.storages.observations.ObservationResultsStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[str] = None)[source]¶

Bases: thoth.storages.result_base.ResultStorageBase

Store results of provenance reports.

RESULT_TYPE = 'provenance-checker'¶

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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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(*args, **kwargs) → 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(*, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, include_end_date: bool = False, only_requests: bool = False) → Generator[str, None, None][source]¶

Get listing of documents available in Ceph as a generator.

Additional parameters can filter results. If start_date is supplied and no end_date is supplied explicitly, the current date is considered as end_date (inclusively).

is_connected() → bool[source]¶: Check if the given database adapter is in connected state.

iterate_results(*, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, include_end_date: bool = False) → Generator[tuple, None, None][source]¶

Iterate over results available in the Ceph.

Additional parameters can filter results. If start_date is supplied and no end_date is supplied explicitly, the current date is considered as end_date (inclusively).

request_exists(document_id: str) → bool[source]¶: Check if a request exists for the given document id.

retrieve_document(document_id: str) → dict[source]¶: Retrieve a document from Ceph by its id.

retrieve_request(document_id: str) → Dict[str, Any][source]¶: Retrieve document capturing requests.

store_document(document: dict, document_id: Optional[str] = None) → str[source]¶: Store the given document in Ceph.

store_file(file_path: str, file_id: str) → str[source]¶: Store the given file in Ceph.

store_request(document_id: str, request: Dict[str, Any]) → str[source]¶

Store the given request.

This function stores a request document for user request traceability.

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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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)[source]¶

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating security-indicators aggregated.

security_indicator_type = 'aggregated'¶

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

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating security-indicators bandit.

security_indicator_type = 'bandit'¶

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

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating with security-indicators cloc.

security_indicator_type = 'cloc'¶

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

Bases: object

Adapter for manipulating 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() → Iterator[str][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: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[str] = None)[source]¶

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting solver results.

RESULT_TYPE = 'solver'¶

get_document_listing(*, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, include_end_date: bool = False, only_requests: bool = False, solver_info: Optional[thoth.storages.solvers._SolverInfo] = None) → Generator[str, None, None][source]¶

Get listing of documents available in Ceph as a generator.

Additional parameters can filter results. If start_date is supplied and no end_date is supplied explicitly, the current date is considered as end_date (inclusively).

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[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync adviser documents into graph.

thoth.storages.sync.sync_analysis_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync image analysis documents into graph.

thoth.storages.sync.sync_dependency_monkey_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync dependency monkey reports into graph database.

thoth.storages.sync.sync_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, 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[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync observations made on Amun into graph database.

thoth.storages.sync.sync_provenance_checker_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync provenance check documents into graph.

thoth.storages.sync.sync_revsolver_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync reverse solver documents into graph.

thoth.storages.sync.sync_security_indicators_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync security indicators results into graph.

thoth.storages.sync.sync_solver_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync solver documents into graph.

Module contents¶

Storage and database adapters for Thoth.

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

Bases: thoth.storages.ceph_cache.CephCache

Adapter for retrieving and storing hashes for cached adviser results.

RESULT_TYPE = 'adviser-cache'¶

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

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting recommendation results.

RESULT_TYPE = 'adviser'¶

class thoth.storages.AnalysesCacheStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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'¶

class thoth.storages.AnalysisByDigest(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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¶

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

Bases: thoth.storages.result_base.ResultStorageBase

Store results of image analyzes.

RESULT_TYPE = 'analysis'¶

class thoth.storages.BuildLogsAnalysesCacheStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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'¶

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

Bases: thoth.storages.result_base.ResultStorageBase

Store results of buildlog parser.

RESULT_TYPE = 'buildlogs-parsed'¶

class thoth.storages.BuildLogsStore(deployment_name=None, *, bucket_prefix: Optional[str] = None, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[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.

class thoth.storages.CephStore(prefix, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[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(prefix_addition: Optional[str] = '') → 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(prefix_addition: Optional[str] = None) → 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.

retrieve_document_attr(object_key: str, attr: str) → Any[source]¶: Retrieve the given attribute of a document 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.

class thoth.storages.DependencyMonkeyReportsStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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() → Iterable[str][source]¶: Iterate over all inspection ids that were run.

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

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting requests to Dependency Monkey runs.

RESULT_TYPE = 'dependency-monkey-requests'¶

class thoth.storages.GraphBackupStore(deployment_name=None, *, host: Optional[str] = None, key_id: Optional[str] = None, secret_key: Optional[str] = None, bucket: Optional[str] = None, region: Optional[str] = None, prefix: Optional[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 .

class thoth.storages.GraphDatabase(engine: Optional[sqlalchemy.engine.base.Engine] = None, sessionmaker=None)[source]¶

Bases: thoth.storages.graph.sql_base.SQLBase

A SQL database adapter providing graph-like operations on top of SQL queries.

DEFAULT_COUNT = 100¶

adviser_document_id_exist(adviser_document_id: str) → bool[source]¶: Check if there is a adviser document record with the given id.

analysis_document_id_exist(analysis_document_id: str) → bool[source]¶: Check if there is an analysis document record with the given id.

analysis_records_exist(analysis_document: dict) → bool[source]¶: Check whether the given analysis document records exist in the graph database.

cache_clear() → None[source]¶: Drop cache of records.

connect() → None[source]¶: Connect to the database.

static construct_connection_string() → str[source]¶: Construct a connection string needed to connect to database.

create_github_app_installation(slug: str, repo_name: str, private: bool, installation_id: str) → bool[source]¶

Create a record for new installation or reactivate uninstalled installation.

Example -: “installation_id”: “236796147”, “repo_name”: “advisor”, “slug”: “thoth-station/advisor”, “private”: False

Accepts installation details passed down by Github. :rtype: True, False :returns True: if installation existed and was updated :returns False: if installation was newly added.

create_hardware_information(hardware: Dict[str, Any], is_external: bool = True) → int[source]¶: Create hardware information in the database.

create_python_cve_record(package_name: str, package_version: str, index_url: str, *, cve_id: str, details: str, link: Optional[str]) → bool[source]¶: Store information about a CVE in the graph database for the given Python package.

create_python_package_version_entity(package_name: str, package_version: Optional[str] = None, index_url: Optional[str] = None, *, only_if_package_seen: bool = False) → Optional[Tuple[thoth.storages.graph.models.PythonPackageVersionEntity, bool]][source]¶

Create a Python package version entity record in the system.

By creating this entity, the system will record and track the given package.

create_python_rule(package_name: str, *, version_specifier: Optional[str] = None, index_url: Optional[str] = None, description: Optional[str] = None) → Dict[str, Any][source]¶: Create the given Python rule.

delete_adviser_result(adviser_document_id: str) → int[source]¶: Delete the corresponding adviser result.

delete_adviser_run(*, end_datetime: Optional[datetime.datetime] = None, adviser_version: Optional[str] = None) → int[source]¶: Delete corresponding adviser data.

delete_analysis_result(analysis_document_id: str) → int[source]¶: Delete the given package-extract entry.

delete_hardware_information(hardware_information_id: int, is_external: bool = True) → None[source]¶: Delete hardware information entry with the given id.

delete_kebechet_github_app_installations(*, id: Optional[int] = None, slug: Optional[str] = None, private: Optional[bool] = None, installation_id: Optional[str] = None, is_active: Optional[bool] = None, runtime_environment_name: Optional[str] = None, info_manager: Optional[bool] = None, pipfile_requirements_manager: Optional[bool] = None, update_manager: Optional[bool] = None, version_manager: Optional[bool] = None, thoth_advise_manager: Optional[bool] = None, thoth_provenance_manager: Optional[bool] = None, last_run_before: Optional[datetime.datetime] = None, last_run_after: Optional[datetime.datetime] = None, external_python_software_stack_id: Optional[int] = None, external_software_environment_id: Optional[int] = None, advised_python_software_stack_id: Optional[int] = None) → int[source]¶

Delete github kebechet installations which match the given filters.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.delete_kebechet_github_app_installations(slug=”foo/bar”, runtime_environment_name=”baz”) [

{
“id”: 0, “slug”: “thoth-station/storages”, “repo_name”: “storages”, “private”: False, “is_active”: True, “info_manager”: True, “pipfile_requirements_manager”: True, …

] :rtype: int :returns: Number of entries which have been deleted

delete_package_extract_run(*, end_datetime: Optional[datetime.datetime] = None, package_extract_version: Optional[str] = None) → int[source]¶: Delete corresponding container image analysis data.

delete_python_package_index(index_url: str) → None[source]¶: Delete the given Python package index.

delete_python_rule(rule_id: int) → int[source]¶: Delete the given Python rule.

delete_solved(*, os_name: str, os_version: str, python_version: str) → int[source]¶: Delete corresponding solver data.

delete_solver_result(solver_document_id: str) → int[source]¶: Delete the corresponding solver result.

dependency_monkey_document_id_exists(dependency_monkey_document_id: str) → bool[source]¶: Check if the given dependency monkey report record exists in the graph database.

drop_all()[source]¶: Drop all content stored in the database.

get_active_kebechet_github_installations_repos() → List[str][source]¶

Get all active repositories names with active Kebechet installation.

Examples:

>>> from thoth.storages import GraphDatabase
>>> graph = GraphDatabase()
>>> graph.get_active_kebechet_github_installations_repos()
['repository_foo_fullname', 'repository_bar_fullname', ...]

get_active_kebechet_github_installations_repos_count_all() → int[source]¶

Return the count of active repos with Kebechet installation.

Example: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.connect() >>> graph.get_active_kebechet_github_installations_repos_count_all() 165

get_adviser_run_count_per_source_type() → Dict[str, int][source]¶

Retrieve number of Adviser run per source type in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_adviser_run_count_per_source_type() {‘GITHUB_APP’: 154, ‘CLI’: 71}

get_adviser_run_document_ids_all(initial_date: Optional[str] = None, final_date: Optional[str] = None, source_type: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100) → List[str][source]¶

Retrieve adviser run document ids.

@params initial_date: DD-MM-YY @params final_date: DD-MM-YY

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_adviser_run_document_ids_all() [‘adviser-343231d’]

get_adviser_run_origins_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → List[str][source]¶

Retrieve all origins (git repos URLs) in Adviser Run.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_adviser_run_origins_all() [’https://github.com/thoth-station/storages’,

‘https://github.com/thoth-station/user-api’, ‘https://github.com/thoth-station/adviser’]

get_alembic_version_count_all() → int[source]¶: Get number of records in alembic version table (1 expected).

get_analysis_metadata(analysis_document_id: str) → Dict[str, Any][source]¶

Get metadata stored for the given analysis document.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_analysis_metadata() {

‘analysis_datetime’: datetime(2019, 10, 7, 18, 57, 22, 658131), ‘analysis_document_id’: ‘package-extract-2ef02c9cea8b1ef7’, ‘package_extract_name’: ‘thoth-package-extract’, ‘package_extract_version’: ‘1.0.1’ }

get_analyzed_image_symbols_all(os_name: str, os_version: str, *, python_version: Optional[str] = None, cuda_version: Optional[str] = None) → List[str][source]¶: Get symbols associated with a given image.

get_analyzer_documents_count_all() → int[source]¶: Get number of image analysis documents synced into graph.

get_bloat_data() → List[Dict[str, Any]][source]¶

Get table bloat data.

documenation: https://raw.githubusercontent.com/pgexperts/pgx_scripts/master/bloat/table_bloat_check.sql It returns data for each table present in the database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_bloat_data() [

{
‘databasename’: ‘postgres’, ‘schemaname’: ‘public’, ‘tablename’: ‘advised’, ‘can_estimate’: True, ‘table_bytes’: 49152.0, ‘table_mb’: 0.047, ‘expected_bytes’: 49152.0, ‘expected_mb’: 0.047, ‘pct_bloat’: 0.0, ‘mb_bloat’: 0.0, ‘est_rows’: 1158.0

}, {

‘databasename’: ‘postgres’, ‘schemaname’: ‘public’, ‘tablename’: ‘adviser_run’, ‘can_estimate’: True, ‘table_bytes’: 720896.0, ‘table_mb’: 0.688, ‘expected_bytes’: 712704.0, ‘expected_mb’: 0.68, ‘pct_bloat’: 1.0, ‘mb_bloat’: 0.01, ‘est_rows’: 3600.0

}, …

]

get_build_software_environment_all(start_offset: int = 0, count: Optional[int] = 100) → List[str][source]¶

Get all software environments available for build.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_run_software_environment_all() [‘quay.io/thoth-station/thoth-pylint:v0.7.0-ubi8’]

get_build_software_environment_analyses_all(build_software_environment_name: str, start_offset: int = 0, count: Optional[int] = 100, convert_datetime: bool = True, is_external: bool = False) → List[dict][source]¶: Get listing of analyses available for the given software environment for build.

get_cve_timestamp() → Optional[datetime.datetime][source]¶: Get CVE timestamp record.

get_database_size() → int[source]¶: Get size of the database in bytes.

get_depends_on(package_name: str, package_version: Optional[str] = None, index_url: Optional[str] = None, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, extras: FrozenSet[Optional[str]] = None, marker_evaluation_result: Optional[bool] = None, is_missing: Optional[bool] = None) → Dict[str, List[Tuple[str, str]]][source]¶

Get dependencies for the given Python package respecting environment and extras.

If no environment is provided, dependencies are returned for all environments as stored in the database.

Extras (as described in PEP-0508) are respected. If no extras is provided (extras=None), all dependencies are returned with all extras specified. A special value of None in extras listing no extra:

extras=frozenset((None,)) - return only dependencies which do not have any extra assigned

extras=frozenset((None, “postgresql”)) - dependencies without extra and with extra “postgresql”

extras=None - return all dependencies (regardless extra)

Environment markers are not taken into account in this query.

get_ecosystem_solver_all() → List[str][source]¶

Get all solvers.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_ecosystem_solver_all() [‘solver-fedora-31-py38’, ‘solver-fedora-32-py37’, ‘solver-fedora-32-py38’, ‘solver-ubi-8-py36’]

get_ecosystem_solver_count_all() → int[source]¶: Get number of solvers.

get_entity_count(entity: Union[sqlalchemy.ext.declarative.api.Base, thoth.storages.graph.models_base.BaseExtension]) → int[source]¶: Get count of a specific entity in the database.

get_error_solved_document_id_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, unsolvable: bool = False, unparseable: bool = False, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, limit_results: bool = True) → List[str][source]¶

Retrieve solver document id with error Python package versions in Thoth Database.

if unsolvable=True -> get_unsolvable_python_package_versions if unparseable=True -> get_unparseable_python_package_versions

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_error_solved_document_id_all() [‘solver-fedora-32-py37-324232’]

get_error_solved_python_package_versions_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, unsolvable: bool = False, unparseable: bool = False, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → List[Tuple[str, str, str]][source]¶

Retrieve solved with error Python package versions in Thoth Database.

if unsolvable=True -> get_unsolvable_python_package_versions if unparseable=True -> get_unparseable_python_package_versions

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_error_solved_python_package_versions_all() [(‘regex’, ‘2018.11.7’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘1.11.0’, ‘https://pypi.org/simple’)]

get_error_solved_python_package_versions_count_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, unsolvable: bool = False, unparseable: bool = False, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → int[source]¶

Retrieve solved with error Python package versions number in Thoth Database.

if unsolvable=True -> get_unsolvable_python_package_versions_count_all if unparseable=True -> get_unparseable_python_package_versions_count_all

get_hardware_environments_all(is_external: bool = False, *, start_offset: int = 0, count: Optional[int] = 100, without_id: bool = True) → List[Dict][source]¶: Get hardware environments (external or internal) registered in the graph database.

get_index_bloat_data() → List[Dict[str, Any]][source]¶

Get index bloat data.

documenation: https://raw.githubusercontent.com/pgexperts/pgx_scripts/master/bloat/index_bloat_check.sql It returns data for each index/table present in the database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_index_bloat_data() [

{
‘database_name’: ‘postgres’, ‘schema_name’: ‘public’, ‘table_name’: ‘depends_on’, ‘index_name’: ‘depends_on_entity_id_idx’, ‘bloat_pct’: 11.0, ‘bloat_mb’: 623.0, ‘index_mb’: 5821.516, ‘table_mb’: 14053.625, ‘index_scans’: 2

}, {

‘database_name’: ‘postgres’, ‘schema_name’: ‘public’, ‘table_name’: ‘python_package_version’, ‘index_name’: ‘python_package_version_package_name_package_version_python__key’, ‘bloat_pct’: 43.0, ‘bloat_mb’: 35.0, ‘index_mb’: 35.0, ‘table_mb’: 111.344, ‘index_scans’: 0

}, …

]

get_index_url_from_id(package_index_id: int) → str[source]¶: Return index URL from id.

get_kebechet_github_app_installations_all(*, id: Optional[int] = None, slug: Optional[str] = None, private: Optional[bool] = None, installation_id: Optional[str] = None, is_active: Optional[bool] = None, runtime_environment_name: Optional[str] = None, info_manager: Optional[bool] = None, pipfile_requirements_manager: Optional[bool] = None, update_manager: Optional[bool] = None, version_manager: Optional[bool] = None, thoth_advise_manager: Optional[bool] = None, thoth_provenance_manager: Optional[bool] = None, last_run_before: Optional[datetime.datetime] = None, last_run_after: Optional[datetime.datetime] = None, external_python_software_stack_id: Optional[int] = None, external_software_environment_id: Optional[int] = None, advised_python_software_stack_id: Optional[int] = None) → List[Dict[str, Any]][source]¶

Get all github kebechet installations with optional filters on values.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_kebechet_github_installations_all() [

{
“id”: 0, “slug”: “thoth-station/storages”, “repo_name”: “storages”, “private”: False, “is_active”: True, “info_manager”: True, “pipfile_requirements_manager”: True, …

] :rtype: List[Dict[str, Any]] :returns: A list of kebechet installations which match the applied filters

get_kebechet_github_installation_info_with_software_environment_all(*, python_version: Optional[str] = None, image_name: Optional[str] = None, image_sha: Optional[str] = None, os_name: Optional[str] = None, os_version: Optional[str] = None, thoth_image_name: Optional[str] = None, thoth_image_version: Optional[str] = None, env_image_name: Optional[str] = None, env_image_tag: Optional[str] = None, cuda_version: Optional[str] = None, environment_type: Optional[str] = None) → List[Dict[str, Any]][source]¶

Get all github kebechet installations with optional filters on software environment.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_kebechet_github_installation_info_with_software_environment_all() [

{
“id”: 0, “slug”: “thoth-station/storages”, “repo_name”: “storages”, “private”: False, “is_active”: True, “info_manager”: True, “pipfile_requirements_manager”: True, …

] :rtype: List[Dict[str, Any]] :returns: A list of kebechet installations which match the applied filters

get_kebechet_github_installations_active_managers(slug: str, runtime_environment_name: Optional[str] = None) → list[source]¶

Return the list of active managers for a particular repository.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_kebechet_github_installations_active_managers(

slug=”thoth-station/srcops-testing”

) [‘thoth_advise_manager’]

get_kebechet_github_installations_active_managers_count_all(kebechet_manager: str, distinct: bool = False) → int[source]¶

Return the number of repos with specific manager active.

Return type: int

:returns Number of active records for registered Kebechet manager.

get_kebechet_github_installations_info_for_python_package_version(package_name: str, *, package_version: Optional[str] = None, index_url: Optional[str] = None, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None) → Dict[str, Dict[str, Any]][source]¶

Return info about repo containing Python Package in the software stack.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_kebechet_github_installations_info_for_python_package_version(

package_name=’click’ index_url=”https://pypi.org/simple”,

)

{

‘thoth-station/jupyter-nbrequirements’:

{: ‘environment_name’: ‘ubi8’, ‘installation_id’: ‘193650988’, ‘private’: False, ‘package_name’: ‘click’, ‘package_version’: ‘7.1.2’, ‘index_url’: ‘https://pypi.org/simple’}

}

get_kebechet_github_installations_software_stack_count_all(*, is_active: Optional[bool] = None)[source]¶: Get number of Kebechet maintained software stacks.

get_last_analysis_datetime() → datetime.datetime[source]¶: Get the datetime of the last container image analysis synced in the database.

get_last_analysis_document_id(thoth_image_name: str, thoth_image_version: str, *, is_external: bool = False) → Optional[Dict[str, str]][source]¶: Get last image analysis (if any) for the given container image.

get_last_solver_datetime(os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None) → datetime.datetime[source]¶: Get the datetime of the last solver run synced in the database.

get_main_table_count() → Dict[str, int][source]¶: Retrieve dictionary mapping main tables to records count.

get_origin_count_per_source_type(distinct: bool = False) → Dict[str, Dict[Tuple[str, str], int]][source]¶

Retrieve number of users of adviser per source type.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_origin_count_per_source_type() {‘KEBECHET’: 48, ‘JUPYTER_NOTEBOOK’: 18, ‘CLI’: 513} >>> graph.get_origin_count_per_source_type(distinct=True) {‘KEBECHET’: 5, ‘JUPYTER_NOTEBOOK’: 2, ‘CLI’: 78}

get_performance_table_count() → Dict[str, int][source]¶: Get dictionary mapping performance tables to records count.

get_pi_component_all() → List[str][source]¶: Retrieve pi components in Thoth database.

get_pi_count(component: str) → Dict[str, int][source]¶: Get dictionary with number of Performance Indicators per type for the PI component selected.

get_python_cve_records_all(package_name: str, package_version: Optional[str] = None) → List[dict][source]¶: Get known vulnerabilities for the given package-version.

get_python_cve_records_count() → int[source]¶: Get number of CVE in Thoth database.

get_python_environment_marker(package_name: str, package_version: str, index_url: str, *, dependency_name: str, dependency_version: str, os_name: str, os_version: str, python_version: str, marker_evaluation_result: Optional[bool] = None) → Optional[str][source]¶

Get Python evaluation marker as per PEP-0508.

@raises NotFoundError: if the given package has no entry in the database

get_python_environment_marker_evaluation_result(package_name: str, package_version: str, index_url: str, *, dependency_name: str, dependency_version: str, os_name: str, os_version: str, python_version: str) → bool[source]¶

Get result of the Python evaluation marker.

The extra part of the environment marker (an exception in PEP-0508) is not taken into account and is substituted with a value which always defaults to True (as it would cause an error during context interpreting). See solver implementation for details.

@raises NotFoundError: if the given package has no entry in the database

get_python_package_hashes_sha256(package_name: str, package_version: str, index_url: str, *, distinct: bool = False) → List[str][source]¶

Get all hashes for Python package in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_hashes_sha256() [

‘9d6863f6c70d034b8c34b3355cb7ba7d2ad799583947265efda41fe67127c23f’, ‘8e4a1f6d89cfaadb486237acbfa24700add01da022dfcf3536e5071d21e13ee0’

]

get_python_package_index_all(enabled: Optional[bool] = None) → List[Dict[str, Any]][source]¶: Get listing of Python package indexes registered in the graph database.

get_python_package_index_urls_all(enabled: Optional[bool] = None) → List[str][source]¶: Retrieve all the URLs of registered Python package indexes.

get_python_package_required_symbols(package_name: str, package_version: str, index_url: str) → List[str][source]¶: Get required symbols for a Python package in a specified version.

get_python_package_version_all(analysis_document_id: str) → List[Dict[str, str]][source]¶: Retrieve Python package information for the given container image analysis.

get_python_package_version_dependents_all(package_name: str, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: int = 100) → List[Dict[str, Any]][source]¶

Get dependents for the given package.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_dependents(“selinon”, os_name=”rhel”, os_version=”8”, python_version=”3.6”) [

{
“index_url”: “https://pypi.org/simple”, “package_name”: “thoth-worker”, “package_version”: “0.0.2”, “version_range”: “>=1.0.0”, “marker_evaluation_result”: True, “marker”: None, “extra”: None,

}

]

get_python_package_version_entities_count_all(*, distinct: bool = False) → int[source]¶: Retrieve number of all Python packages in Thoth Database.

get_python_package_version_entities_names_all() → List[str][source]¶: Retrieve names of Python package entities in the Thoth’s knowledge base.

get_python_package_version_import_packages_all(import_name: str, distinct: bool = False) → List[Dict[str, str]][source]¶

Retrieve Python package name for the given import package name.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_version_import_packages_all(“faust.*”) [

{‘import’: ‘faust.web.apps’,
‘index_url’: ‘https://pypi.org/simple’, ‘package_name’: ‘faust’, ‘package_version’: ‘1.9.0’},

{‘import’: ‘faust.web.cache’,
‘index_url’: ‘https://pypi.org/simple’, ‘package_name’: ‘faust’, ‘package_version’: ‘1.9.0’},

{‘import’: ‘faust.web.cache.backends’,
‘index_url’: ‘https://pypi.org/simple’, ‘package_name’: ‘faust’, ‘package_version’: ‘1.9.0’},

{‘import’: ‘faust.web.drivers’,
‘index_url’: ‘https://pypi.org/simple’, ‘package_name’: ‘faust’, ‘package_version’: ‘1.9.0’} …

]

get_python_package_version_metadata(package_name: str, package_version: str, index_url: str) → Dict[str, str][source]¶: Retrieve Python package metadata.

get_python_package_version_names_all(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, sort: bool = False, like: Optional[str] = None) → List[str][source]¶

Retrieve names of Python Packages known by Thoth.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_packages_names_all() [‘regex’, ‘tensorflow’]

get_python_package_version_names_count_all(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, like: Optional[str] = None) → int[source]¶

Retrieve names of Python Packages known by Thoth.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_packages_names_count_all() 156468

static get_python_package_version_platform_all() → List[str][source]¶

Retrieve all platforms stored in the database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_version_platform_all() [‘linux-x86_64’]

get_python_package_version_records(package_name: str, package_version: str, index_url: Optional[str], *, os_name: Optional[str], os_version: Optional[str], python_version: Optional[str]) → List[dict][source]¶: Get records for the given package regardless of index_url.

get_python_package_version_solver_rules_all(package_name: str, package_version: Optional[str] = None, index_url: Optional[str] = None) → List[Tuple[int, Optional[str], Optional[str], str]][source]¶: Get rules assigned for the given Python package.

get_python_package_version_trove_classifiers_all(package_name: str, package_version: str, index_url: str, *, os_name: str, os_version: str, python_version: str) → List[str][source]¶

Get Python trove classifiers.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_version_trove_classifiers_all( “cowsay”, “4.0”, “https://pypi.org/simple”, os_name=”rhel”, os_version=”9”, python_version=”3.9”) [“PROGRAMMING LANGUAGE :: PYTHON :: 3.9”, “OPERATING SYSTEM :: OS INDEPENDENT”]

get_python_package_versions_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, is_missing: Optional[bool] = None) → List[Tuple[str, str, str]][source]¶

Retrieve Python package versions in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_versions_all() [(‘regex’, ‘2018.11.7’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘1.11.0’, ‘https://pypi.org/simple’)]

get_python_package_versions_all_count(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, sort_by: Optional[thoth.storages.graph.enums.QuerySortTypeEnum] = None) → thoth.storages.graph.query_result_base.PythonQueryResult[source]¶

Retrieve number of versions per Python package name in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_versions_all_count() {‘setuptools’: 988, ‘pip’: 211, ‘termcolor’: 14, ‘six’: 42}

get_python_package_versions_count(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[Tuple[str, str, str], int][source]¶

Retrieve number of Python Package (package_name, package_version, index_url) in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_versions_count() {(‘absl-py’, ‘0.1.10’, ‘https://pypi.org/simple’): 1, (‘absl-py’, ‘0.2.1’, ‘https://pypi.org/simple’): 1}

get_python_package_versions_count_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, is_missing: Optional[bool] = None) → int[source]¶: Retrieve Python package versions number in Thoth Database.

get_python_package_versions_count_per_index(index_url: str, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, Dict[Tuple[str, str], int]][source]¶

Retrieve number of Python package versions per index url in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_versions_count_per_index(index_url=’https://pypi.org/simple’) {’https://pypi.org/simple’: {(‘absl-py’, ‘0.1.10’): 1, (‘absl-py’, ‘0.2.1’): 1}}

get_python_package_versions_count_per_version(package_name: str, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, Dict[str, int]][source]¶

Retrieve number of Python package versions per index url in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_package_versions_count_per_version(package_name=’tensorflow’) {‘1.14.0rc0’: {’https://pypi.org/simple’: 1}, ‘1.13.0rc2’: {’https://pypi.org/simple’: 1}}

get_python_package_versions_per_index(index_url: str, *, distinct: bool = False) → Dict[str, List[str]][source]¶: Retrieve listing of Python packages (solved) known to graph database instance for the given index.

get_python_packages_all(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → List[Tuple[str, str]][source]¶

Retrieve Python packages with index in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_packages_all() [(‘regex’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘https://pypi.org/simple’)]

get_python_packages_all_versions(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, List[Tuple[str, str]]][source]¶

Retrieve Python package versions per package in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_python_packages_all_versions() {‘absl-py’: [(‘0.1.10’, ‘https://pypi.org/simple’), (‘0.2.1’, ‘https://pypi.org/simple’)]}

get_python_packages_count_all(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → int[source]¶: Retrieve number of versions per Python package in Thoth Database.

get_python_rule(rule_id: int) → Dict[str, Any][source]¶: Get the given Python rule.

get_python_rule_all(*, package_name: Optional[str] = None, index_url: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100) → List[Dict[str, Any]][source]¶: Get all the Python rules matching the given query criteria.

get_python_software_stack_count_all(is_external: bool = False, software_stack_type: Optional[str] = None, distinct: bool = False) → int[source]¶: Get number of Python software stacks available filtered by type.

get_relation_table_count() → Dict[str, int][source]¶: Retrieve dictionary mapping relation tables to records count.

get_rpm_package_version_all(analysis_document_id: str) → List[Dict[str, str]][source]¶: Retrieve RPM package information for the given container image analysis.

get_run_software_environment_all(start_offset: int = 0, count: Optional[int] = 100, is_external: bool = False) → List[str][source]¶

Get all software environments available for run.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_run_software_environment_all() [‘quay.io/thoth-station/thoth-pylint:v0.7.0-ubi8’]

get_run_software_environment_analyses_all(run_software_environment_name: str, start_offset: int = 0, count: Optional[int] = 100, convert_datetime: bool = True, is_external: bool = False) → List[dict][source]¶

Get listing of analyses available for the given software environment for run.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_run_software_environment_analyses_all() [{

‘analysis_datetime’: datetime(2019, 10, 7, 18, 57, 22, 658131), ‘analysis_document_id’: ‘package-extract-2ef02c9cea8b1ef7’, ‘package_extract_name’: ‘thoth-package-extract’, ‘package_extract_version’: ‘1.0.1’ }]

get_script_alembic_version_head() → str[source]¶: Get alembic version head from alembic folder scripts.

get_si_aggregated_python_package_version(package_name: str, package_version: str, index_url: str) → Dict[str, int][source]¶

Get Aggregate Security Indicators (SI) results per Python package version.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_si_aggregated_python_package_version(

package_name=’thoth-common’, package_version=’0.10.0’, index_url=’https://pypi.org/simple’

)

{: ‘severity_high_confidence_high’: 0, ‘severity_high_confidence_low’: 0, ‘severity_high_confidence_medium’: 0, ‘severity_high_confidence_undefined’: 0, ‘severity_low_confidence_high’: 0, ‘severity_low_confidence_low’: 0, ‘severity_low_confidence_medium’: 0, ‘severity_low_confidence_undefined’: 0, ‘severity_medium_confidence_high’: 0, ‘severity_medium_confidence_low’: 0, ‘severity_medium_confidence_medium’: 0, ‘severity_medium_confidence_undefined’: 0, ‘number_of_analyzed_files’: 0, ‘number_of_files_total’: 0, ‘number_of_files_with_severities’: 0, ‘number_of_filtered_files’: 0, ‘number_of_python_files’: 39, ‘number_of_lines_with_comments_in_python_files’: 922, ‘number_of_blank_lines_in_python_files’: 2760, ‘number_of_lines_with_code_in_python_files’: 9509, ‘total_number_of_files’: 75, ‘total_number_of_lines’: 36856, ‘total_number_of_lines_with_comments’: 2895, ‘total_number_of_blank_lines’: 6737, ‘total_number_of_lines_with_code’: 27224

}

get_si_analyzed_python_package_versions_all(*, distinct: bool = False) → List[Tuple[str, str, str]][source]¶

Get SI analyzed Python package versions in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_si_analyzed_python_package_versions_all() [(‘fbprophet’, ‘0.4’, ‘https://pypi.org/simple’)]

get_si_analyzed_python_package_versions_count_all(*, distinct: bool = False) → int[source]¶: Get SI analyzed Python package versions number in Thoth Database.

get_si_unanalyzed_python_package_versions_all(start_offset: int = 0, count: Optional[int] = 100, distinct: bool = True, randomize: bool = True, provides_source_distro: bool = True, si_error: bool = False) → List[Tuple[str, str, str]][source]¶

Retrieve solved Python package versions in Thoth Database, that are not anaylyzed by SI.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_si_unanalyzed_python_package_versions_all() [(‘crossbar’, ‘0.10.0’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘1.11.0’, ‘https://pypi.org/simple’)]

get_si_unanalyzed_python_package_versions_count_all(index_url: Optional[str] = None, *, distinct: bool = False, provides_source_distro: bool = True, si_error: bool = False) → int[source]¶: Get SI unanalyzed Python package versions number in Thoth Database.

get_software_environments_all(is_external: bool = False, convert_datetime: bool = True, *, start_offset: int = 0, count: Optional[int] = 100, env_image_name: Optional[str] = None, env_image_tag: Optional[str] = None, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, cuda_version: Optional[str] = None, image_name: Optional[str] = None, library_name: Optional[str] = None, symbol: Optional[str] = None, package_name: Optional[str] = None, rpm_package_name: Optional[str] = None) → List[Dict][source]¶

Get software environments (external or internal) registered in the graph database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_software_environments_all()

[

{: ‘cuda_version’: None, ‘datetime’: datetime.datetime(2021, 12, 15, 19, 27, 52, 803266), ‘env_image_name’: None, ‘env_image_tag’: None, ‘environment_name’: ‘quay.io/thoth-station/s2i-thoth-ubi8-py39:v0.32.3’, ‘environment_type’: ‘RUNTIME’, ‘image_sha’: ‘bcf4fa447e5dc889015afb4d3c54ef4a3aaddc3260fce072311602df34ffac3c’, ‘os_name’: ‘rhel’, ‘os_version’: ‘8’, ‘package_extract_document_id’: ‘package-extract-211215162259-33c8d9c730b775eb’, ‘python_version’: ‘3.9’, ‘thoth_image_name’: ‘quay.io/thoth-station/s2i-thoth-ubi8-py39’, ‘thoth_image_version’: ‘0.32.3’ }

]

get_software_environments_count_all(is_external: bool = False, *, env_image_name: Optional[str] = None, env_image_tag: Optional[str] = None, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, cuda_version: Optional[str] = None, image_name: Optional[str] = None, library_name: Optional[str] = None, symbol: Optional[str] = None, package_name: Optional[str] = None, rpm_package_name: Optional[str] = None) → int[source]¶: Get number of software environments stored.

get_solved_python_package_version_environments_all(package_name: str, package_version: str, index_url: str, *, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → List[Dict[str, str]][source]¶

Retrieve all the environments that were used to solve the given package.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_package_version_environments_all( package_name=”flask”, package_version=”2.0.2”, index_url=”https://pypi.org/simple”) [

{“os_name”: “rhel”, “os_version”: “8”, “python_version”: “3.8”}, {“os_name”: “fedora”, “os_version”: “35”, “python_version”: “3.9”},

]

get_solved_python_package_versions_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, is_missing: Optional[bool] = None) → List[Tuple[str, str, str]][source]¶

Retrieve solved Python package versions in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_package_versions_all() [(‘regex’, ‘2018.11.7’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘1.11.0’, ‘https://pypi.org/simple’)]

get_solved_python_package_versions_count(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[Tuple[str, str, str], int][source]¶

Retrieve number of Python Package (package_name, package_version, index_url) solved in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_package_versions_count() {(‘absl-py’, ‘0.1.10’, ‘https://pypi.org/simple’): 1, (‘absl-py’, ‘0.2.1’, ‘https://pypi.org/simple’): 1}

get_solved_python_package_versions_count_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, is_missing: Optional[bool] = None) → int[source]¶: Retrieve solved Python package versions number in Thoth Database.

get_solved_python_package_versions_count_per_index(index_url: str, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, Dict[Tuple[str, str], int]][source]¶

Retrieve number of solved Python package versions per index url in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_package_versions_count_per_index(index_url=’https://pypi.org/simple’) {’https://pypi.org/simple’: {(‘absl-py’, ‘0.1.10’): 1, (‘absl-py’, ‘0.2.1’): 1}}

get_solved_python_package_versions_count_per_version(package_name: str, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, Dict[str, int]][source]¶

Retrieve number of solved Python package versions per package version in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_package_versions_count_per_version(package_name=’tensorflow’) {‘1.14.0rc0’: {’https://pypi.org/simple’: 1}, ‘1.13.0rc2’: {’https://pypi.org/simple’: 1}}

get_solved_python_package_versions_software_environment_all() → List[Dict[str, str]][source]¶: Retrieve software environment configurations used to solve Python packages.

get_solved_python_packages_all(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → List[Tuple[str, str]][source]¶

Retrieve solved Python package with index in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_packages_all() [(‘regex’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘https://pypi.org/simple’)]

get_solved_python_packages_all_versions(*, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, List[Tuple[str, str]]][source]¶

Retrieve solved Python package versions per package in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solved_python_packages_all_versions() {‘absl-py’: [(‘0.1.10’, ‘https://pypi.org/simple’), (‘0.2.1’, ‘https://pypi.org/simple’)]}

get_solved_python_packages_count_all(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → int[source]¶: Retrieve number of solved Python package versions in Thoth Database.

get_solver_document_id_all(package_name: str, package_version: Optional[str] = None, index_url: Optional[str] = None, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, sort: bool = False) → List[str][source]¶

Get the solver document specific to the given package that has been solved.

If sorted, the latest solver document is at index 0.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.graph.get_solver_document_id_all(“selinon”, “1.0.0”, “https://pypi.org/simple”, os_name=”rhel”, os_version=”8”, python_version=”3.8”, sort=True) [

“solver-rhel-8-py38-210713010253-a24759ebdaa1442c”, “solver-rhel-8-py38-210711003643-187580eee457bb3f”, “solver-rhel-8-py38-210707005412-4965667a05a39006”

]

get_solver_documents_count_all() → int[source]¶: Get number of solver documents synced into graph.

get_solver_run_document_ids_all(initial_date: Optional[str] = None, final_date: Optional[str] = None, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, has_error: bool = False, unsolvable: bool = False, unparseable: bool = False) → List[str][source]¶

Retrieve solver run document ids.

@params initial_date: DD-MM-YY @params final_date: DD-MM-YY

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_solver_run_document_ids_all() [‘solver-rhel-8-py38-343231d’]

get_table_alembic_version_head() → str[source]¶: Get alembic version head from database table.

get_thoth_s2i_all(is_external: bool = False) → List[Tuple[str, str]][source]¶: Get all the Thoth s2i container images available.

get_thoth_s2i_analyzed_image_symbols_all(thoth_image_name: str, thoth_image_version: str, is_external: bool = False) → List[str][source]¶: Get symbols associated with a given Thoth s2i container image.

get_thoth_s2i_package_extract_analysis_document_id_all(thoth_image_name: str, thoth_image_version: str, is_external: bool = False) → List[str][source]¶: Get package-extract analysis ids for the given Thoth s2i.

get_unsolved_python_package_versions_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False, randomize: bool = True) → List[Tuple[str, Optional[str], Optional[str]]][source]¶

Retrieve unsolved Python package versions in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_package_versions_all() [(‘regex’, ‘2018.11.7’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘1.11.0’, ‘https://pypi.org/simple’)]

get_unsolved_python_package_versions_count(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → Dict[Tuple[str, str, str], int][source]¶

Retrieve number of unsolved versions per Python package in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_package_versions_count() {(‘absl-py’, ‘0.1.10’, ‘https://pypi.org/simple’): 1, (‘absl-py’, ‘0.2.1’, ‘https://pypi.org/simple’): 1}

get_unsolved_python_package_versions_count_all(package_name: Optional[str] = None, package_version: Optional[str] = None, index_url: Optional[str] = None, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → int[source]¶: Retrieve unsolved Python package versions number in Thoth Database.

get_unsolved_python_package_versions_count_per_index(index_url: str, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → Dict[str, Dict[Tuple[str, str], int]][source]¶

Retrieve number of unsolved Python package versions per index url in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_package_versions_count_per_index(index_url=’https://pypi.org/simple’) {’https://pypi.org/simple’: {(‘absl-py’, ‘0.1.10’): 1, (‘absl-py’, ‘0.2.1’): 1}}

get_unsolved_python_package_versions_count_per_version(package_name: str, *, start_offset: int = 0, count: Optional[int] = 100, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, distinct: bool = False) → Dict[str, Dict[str, int]][source]¶

Retrieve number of unsolved Python package versions per package version in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_package_versions_count_per_version(package_name=’tensorflow’) {‘1.14.0rc0’: {’https://pypi.org/simple’: 1}, ‘1.13.0rc2’: {’https://pypi.org/simple’: 1}}

get_unsolved_python_packages_all(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → List[Tuple[str, Optional[str]]][source]¶

Retrieve unsolved Python package with index in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_packages_all() [(‘regex’, ‘https://pypi.org/simple’), (‘tensorflow’, ‘https://pypi.org/simple’)]

get_unsolved_python_packages_all_per_adviser_run(source_type: str) → Dict[str, List[str]][source]¶

Retrieve all unsolved packages for a certain Adviser Run that need to be re run.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_packages_all_per_adviser_run() {‘adviser-04ab56d6’: [‘black’], ‘adviser-054ab56d6’: [‘black’, ‘numpy’]}

get_unsolved_python_packages_all_versions(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, start_offset: int = 0, count: Optional[int] = 100, distinct: bool = False) → Dict[str, List[Tuple[str, str]]][source]¶

Retrieve unsolved Python package versions per package in Thoth Database.

Examples: >>> from thoth.storages import GraphDatabase >>> graph = GraphDatabase() >>> graph.get_unsolved_python_packages_all_versions() {‘absl-py’: [(‘0.1.10’, ‘https://pypi.org/simple’), (‘0.2.1’, ‘https://pypi.org/simple’)]}

has_python_solver_error(package_name: str, package_version: str, index_url: str, *, os_name: Optional[str], os_version: Optional[str], python_version: Optional[str]) → bool[source]¶: Retrieve information whether the given package has any solver error.

initialize_schema()[source]¶: Initialize schema of database.

inspection_document_id_exist(inspection_document_id: str) → bool[source]¶: Check if there is an inspection document record with the given id.

inspection_document_id_result_number_exists(inspection_document_id: str, inspection_result_number: int) → bool[source]¶: Check if the given inspection id result number record exists in the graph database.

is_database_corrupted() → bool[source]¶

Run the amcheck extension to check for DB corruption, false negatives are possible but false positives are not.

amcheck documenation: https://www.postgresql.org/docs/10/amcheck.html

bool: False: No DB corruption detected by amcheck (not definitive) True: Database is corrupted

is_python_package_index_enabled(url: str) → bool[source]¶: Check if the given Python package index is enabled.

is_python_package_version_is_missing(package_name: str, package_version: str, index_url: str) → bool[source]¶: Check whether is_missing flag is set for python package version.

is_schema_up2date() → bool[source]¶: Check if the current schema is up2date with the one configured on database side.

static normalize_python_index_url(index_url: Optional[str]) → Optional[str][source]¶: Map python index url.

static normalize_python_package_name(package_name: str) → str[source]¶: Normalize Python package name based on PEP-0503.

static normalize_python_package_version(package_version: str) → str[source]¶: Normalize Python package name based on PEP-440.

provenance_checker_document_id_exist(provenance_checker_document_id: str) → bool[source]¶: Check if there is a provenance-checker document record with the given id.

purge_adviser_documents(*, end_datetime: Optional[datetime.datetime] = None, adviser_version: Optional[str] = None) → int[source]¶: Store and purge to be deleted adviser documents to Ceph.

purge_package_extract_documents(*, end_datetime: Optional[datetime.datetime] = None, package_extract_version: Optional[str] = None) → int[source]¶: Store and purge to be deleted package extract documents to Ceph.

purge_solver_documents(*, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None) → int[source]¶: Store and purge to be deleted solver documents to Ceph.

python_package_exists(package_name: str) → bool[source]¶: Check if the given Python package exists regardless of version.

static python_package_version_depends_on_platform_exists(platform: str) → bool[source]¶: Check if the given platform has some records in the database.

python_package_version_exists(package_name: str, package_version: str, index_url: Optional[str] = None, solver_name: Optional[str] = None) → bool[source]¶

Check if the given Python package version exists in the graph database.

If optional solver_name parameter is set, the call answers if the given package was solved by the given solver. Otherwise, any solver run is taken into account.

register_python_package_index(url: str, warehouse_api_url: Optional[str] = None, verify_ssl: bool = True, enabled: bool = False, only_if_package_seen: bool = True) → bool[source]¶: Register the given Python package index in the graph database.

retrieve_dependent_packages(package_name: str, package_version: Optional[str] = None) → Dict[str, List[str]][source]¶: Get mapping package name to package version of packages that depend on the given package.

retrieve_transitive_dependencies_python(package_name: str, package_version: str, index_url: str, *, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None, extras: Optional[FrozenSet[Optional[str]]] = None, marker_evaluation_result: Optional[bool] = None) → List[Tuple[Tuple[str, str, str], Tuple[str, str, str], Tuple[str, str, str], Tuple[str, str, str], Optional[Tuple[str, str, str]], Optional[Tuple[str, str, str]]]][source]¶

Get all transitive dependencies for the given package by traversing dependency graph.

It’s much faster to retrieve just dependency ids for the transitive dependencies as most of the time is otherwise spent in serialization and deserialization of query results. The ids are obtained later on (kept in ids map, see bellow).

The ids map represents a map to optimize number of retrievals - not to perform duplicate queries into graph instance.

Extras are taken into account only for direct dependencies. Any extras required in libraries used in transitive dependencies are not required as solver directly report dependencies regardless extras configuration - see get_depends_on docs for extras parameter values..

retrieve_transitive_dependencies_python_multi(*package_tuples, os_name: Optional[str] = None, os_version: Optional[str] = None, python_version: Optional[str] = None) → Dict[Tuple[str, str, str], Set[Tuple[Tuple[str, str, str], Tuple[str, str, str], Tuple[str, str, str], Tuple[str, str, str], Optional[Tuple[str, str, str]], Optional[Tuple[str, str, str]]]]][source]¶: Get all transitive dependencies for a given set of packages by traversing the dependency graph.

set_cve_timestamp(timestamp: datetime.datetime) → None[source]¶: Set CVE timestamp record.

set_python_package_index_state(url: str, *, enabled: bool) → None[source]¶: Enable or disable Python package index.

si_aggregated_document_id_exists(si_aggregated_run_document_id: str) → bool[source]¶: Check if the given security indicator aggregated report record exists in the graph database.

si_aggregated_python_package_version_exists(package_name: str, package_version: str, index_url: str) → bool[source]¶: Check if Aggregate Security Indicators (SI) results exists for Python package version.

solved_software_environment_exists(os_name: str, os_version: str, python_version: str) → bool[source]¶: Check if there are any solved packages for the given software environment.

solver_document_id_exists(solver_document_id: str) → bool[source]¶: Check if there is a solver document record with the given id.

solver_records_exist(solver_document: dict) → bool[source]¶: Check if the given solver document record exists.

stats() → dict[source]¶: Get statistics for this adapter.

sync_adviser_result(document: dict) → None[source]¶: Sync adviser result into graph database.

sync_analysis_result(document: dict) → None[source]¶: Sync the given analysis result to the graph database.

sync_dependency_monkey_result(document: dict) → None[source]¶: Sync reports of dependency monkey runs.

sync_inspection_result(document) → None[source]¶: Sync the given inspection document into the graph database.

sync_provenance_checker_result(document: dict) → None[source]¶: Sync provenance checker results into graph database.

sync_revsolver_result(document: Dict[str, Any]) → None[source]¶

Sync results of the reverse solver.

This updates relations for DependsOn on a new package release.

sync_security_indicator_aggregated_result(document: dict) → None[source]¶: Sync the given security-indicator aggregated result to the graph database.

sync_solver_result(document: dict, *, force: bool = False) → None[source]¶: Sync the given solver result to the graph database.

update_kebechet_github_installations_on_is_active(slug: str) → bool[source]¶

Deactivate the app on getting an uninstall event.

Passed a slug name to be deactivated. Example - slug:’thoth-station/advisor’ :rtype: True, False :returns True: if installation existed and was deactivated. :returns False: if installation was not found.

update_kebechet_installation_using_files(slug: str, installation_id: str, requirements: Optional[dict] = None, requirements_lock: Optional[dict] = None, thoth_config: Optional[dict] = None, runtime_environment_name: Optional[str] = None, private: bool = False)[source]¶: Update info about kebechet installation.

update_missing_flag_package_version(package_name: str, package_version: str, index_url: str, value: bool) → None[source]¶: Update value of is_missing flag for PythonPackageVersion.

update_provides_source_distro_package_version(package_name: str, package_version: str, index_url: str, value: bool) → None[source]¶: Update value of is_si_analyzable flag for PythonPackageVersion.

update_python_package_hash_present_flag(package_name: str, package_version: str, index_url: str, sha256_hash: str)[source]¶: Remove hash associated with python package in the graph.

class thoth.storages.InspectionBuildsStore(inspection_id: str)[source]¶

Bases: thoth.storages.inspections._InspectionBase

An adapter for retrieving inspection builds.

ceph¶

inspection_id¶

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() → Dict[str, Any][source]¶: Retrieve specification used for the build, captures also run specification.

sub_prefix = 'build'¶

class thoth.storages.InspectionResultsStore(inspection_id: str)[source]¶

Bases: thoth.storages.inspections._InspectionBase

An adapter for manipulating with inspection results.

ceph¶

classmethod get_document_id(document: Dict[str, 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.

inspection_id¶

iter_inspection_results() → Generator[Dict[str, Any], None, None][source]¶: Iterate over inspection results.

retrieve_hwinfo(item: int) → Dict[str, 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) → Dict[str, Any][source]¶: Obtain the actual result for the given inspection run.

sub_prefix = 'results'¶

class thoth.storages.InspectionStore(inspection_id: str)[source]¶

Bases: object

Adapter for manipulating 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() → Generator[str, None, None][source]¶: Iterate over inspection ids stored.

results¶

retrieve_specification() → Dict[str, Any][source]¶: Retrieve specification used for this inspection.

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

Bases: thoth.storages.ceph_cache.CephCache

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

RESULT_TYPE = 'provenance-cache'¶

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

Bases: thoth.storages.result_base.ResultStorageBase

Store results of provenance reports.

RESULT_TYPE = 'provenance-checker'¶

class thoth.storages.SIAggregatedStore(security_indicator_id: str)[source]¶

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating security-indicators aggregated.

security_indicator_type = 'aggregated'¶

class thoth.storages.SIBanditStore(security_indicator_id: str)[source]¶

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating security-indicators bandit.

security_indicator_type = 'bandit'¶

class thoth.storages.SIClocStore(security_indicator_id: str)[source]¶

Bases: thoth.storages.security_indicators._SecurityIndicatorBase

An adapter for manipulating with security-indicators cloc.

security_indicator_type = 'cloc'¶

class thoth.storages.SecurityIndicatorsResultsStore(security_indicator_id: str)[source]¶

Bases: object

Adapter for manipulating 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() → Iterator[str][source]¶: Iterate over security_indicators ids stored.

security_indicator_id¶

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

Bases: thoth.storages.result_base.ResultStorageBase

Adapter for persisting solver results.

RESULT_TYPE = 'solver'¶

get_document_listing(*, start_date: Optional[datetime.date] = None, end_date: Optional[datetime.date] = None, include_end_date: bool = False, only_requests: bool = False, solver_info: Optional[thoth.storages.solvers._SolverInfo] = None) → Generator[str, None, None][source]¶

Get listing of documents available in Ceph as a generator.

Additional parameters can filter results. If start_date is supplied and no end_date is supplied explicitly, the current date is considered as end_date (inclusively).

static get_solver_name_from_document_id(solver_document_id: str) → str[source]¶: Retrieve solver name from solver’s document id.

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

Bases: thoth.storages.base.StorageBase

Access logs stored by Argo Workflows.

connect() → None[source]¶: Connect to Ceph.

get_log(workflow_id: str) → str[source]¶: Obtain log from the given workflow.

thoth.storages.sync_adviser_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync adviser documents into graph.

thoth.storages.sync_analysis_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync image analysis documents into graph.

thoth.storages.sync_dependency_monkey_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync dependency monkey reports into graph database.

thoth.storages.sync_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, 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_inspection_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync observations made on Amun into graph database.

thoth.storages.sync_provenance_checker_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync provenance check documents into graph.

thoth.storages.sync_security_indicators_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync security indicators results into graph.

thoth.storages.sync_solver_documents(document_ids: Optional[Iterable[str]] = None, force: bool = False, graceful: bool = False, graph: Optional[thoth.storages.graph.postgres.GraphDatabase] = None, is_local: bool = False) → Tuple[int, int, int, int][source]¶: Sync solver documents into graph.

thoth.storages package¶

Subpackages¶

Submodules¶

thoth.storages.advisers module¶

thoth.storages.advisers_cache module¶

thoth.storages.analyses module¶

thoth.storages.analyses_by_digest module¶

thoth.storages.analyses_cache module¶

thoth.storages.base module¶

thoth.storages.buildlogs module¶

thoth.storages.buildlogs_analyses_cache module¶

thoth.storages.buildlogs_parsed module¶

thoth.storages.ceph module¶

thoth.storages.ceph_cache module¶

thoth.storages.cli module¶

thoth.storages.dependency_monkey_reports module¶

thoth.storages.dependency_monkey_requests module¶

thoth.storages.exceptions module¶

thoth.storages.graph_backup module¶

thoth.storages.inspections module¶

thoth.storages.logs module¶

thoth.storages.observations module¶

thoth.storages.provenance module¶

thoth.storages.provenance_cache module¶

thoth.storages.result_base module¶

thoth.storages.result_schema module¶

thoth.storages.revsolvers module¶

thoth.storages.security_indicators module¶

thoth.storages.solvers module¶

thoth.storages.sync module¶

Module contents¶

Table of Contents

Related Topics

This Page