vdx_helper module

class vdx_helper.vdx_helper.VDXHelper(api_url: str, auth_url: str, client_secret: str, client_id: str)

Bases: object

Helper class that allows its users to connect to the VDX Core API with little fuss. Each method corresponds to an endpoint on the Core API, including all of its parameters. By default, the results are mapped into specific objects, but a JSON mapper can also be used (directly returns the results in JSON), as well as custom mappers. More information on mappers on the vdx_helper.mappers module.

To use, create an instance of this class by providing the required information, and then invoke any of the available methods. The authentication flow is fully dealt with by this class, as long as the correct client secret and URL is provided.

Parameters:

api_url (str) – The url for the VDX Core API. For example, https://vizidox.com/api
auth_url (str) – The url for the VDX Core API authentication server. For example, https://vizidox.com/auth
client_secret (str) – The client secret for authentication
client_id (str) – The client ID for authentication

create_credential(title: str, metadata: ~typing.Dict[str, ~typing.Any], tags: ~typing.Optional[~typing.Iterable[str]] = None, file_hashes: ~typing.Optional[~typing.List[str]] = None, cred_ids: ~typing.Optional[~typing.List[~uuid.UUID]] = None, expiry_date: ~typing.Optional[str] = None, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function credential_mapper>) → Any

Create a new credential from a previously uploaded file, or provided metadata.

Parameters:

title (str) – The title of the credential
metadata (Dict[str, Any]) – Metadata values to filter by. A dictionary is expected with the key corresponding to the field name found in the metadata, and the value corresponding to that field’s value. Can be an empty dictionary as long as at least one file hash is provided.
tags (Iterable[str], optional) – Optional text tags that can be used to identify and filter the credential. Must have at least 3 characters, and can only contain alphanumeric characters, ‘-’ or ‘_’
file_hashes (List[str], optional) – List of hashes of files to associate to the credential. Must be hashes of files previously uploaded to the core API. Can be None as long as provided metadata is not empty.
cred_ids (List[uuid.UUID], optional) – List of credentials to associate to the new credential.
expiry_date (datetime.datetime, optional) – Date the credential should expire, if applicable
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

delete_credential_tag(cred_uid: UUID, tag: str) → None

Deletes a specific tag from the given credential.

Parameters:

cred_uid (uuid.UUID) – UUID of the credential to update
tag (str) – Tag to delete

download_certificate(cert_uid: UUID) → BytesIO

Downloads the JSON proof file for the Certificate, containing all relevant Blockchain details as well as the metadata of the corresponding issued credential.

Parameters:: cert_uid (uuid.UUID) – UUID of the certificate to download
Returns:: The result of the endpoint call
Return type:: io.BytesIO

get_certificates(mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, *, job_uid: ~typing.Optional[~uuid.UUID] = None, cred_uid: ~typing.Optional[~uuid.UUID] = None, uid: ~typing.Optional[~uuid.UUID] = None, start_date: ~typing.Optional[~datetime.datetime] = None, end_date: ~typing.Optional[~datetime.datetime] = None, and_credential_tags: ~typing.Optional[str] = None, or_credential_tags: ~typing.Optional[str] = None, and_job_tags: ~typing.Optional[str] = None, or_job_tags: ~typing.Optional[str] = None, verification_status: ~typing.Optional[str] = None, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Retrieve all partner certificates, or filter by job, credential, uid, issued date (start_date < issued_date < end_date), and credential or job tags.

More information on tags on the VDX Core API official documentation.

Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
job_uid (uuid.UUID, optional) – Job UID to filter by
cred_uid – Credential UID to filter by
uid (uuid.UUID, optional) – Certificate UID to filter by
start_date (datetime.datetime, optional) – Filter by Certificates issued after the given start_date
end_date (datetime.datetime, optional) – Filter by Certificates issued before the given end_date
and_credential_tags (List[str], optional) – Filter Certificates of Credentials that contain all the given tags
or_credential_tags (List[str], optional) – Filter Certificates of Credentials that contain at least one of the given tags
and_job_tags (List[str], optional) – Filter Certificates issued on Jobs that contain all the given tags
or_job_tags (List[str], optional) – Filter Certificates issued on Jobs that contain at least one of the given tags
verification_status (str, optional) – Filter Certificates in the given verification status
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

typing.Any

get_credential(cred_uid: ~uuid.UUID, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function credential_mapper>) → Any

Retrieve a specific credential from the Core API, via its UUID.

Parameters:

cred_uid (uuid.UUID) – Credential UID to obtain
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

get_credentials(mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, *, metadata: ~typing.Optional[~typing.Dict[str, ~typing.Any]] = None, uid: ~typing.Optional[~uuid.UUID] = None, start_date: ~typing.Optional[~datetime.datetime] = None, end_date: ~typing.Optional[~datetime.datetime] = None, and_tags: ~typing.Optional[str] = None, or_tags: ~typing.Optional[str] = None, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Retrieve all partner credentials, or filter by metadata values, uid, upload_date (start_date < upload_date < end_date), and tags.

More information on tags on the VDX Core API official documentation. Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
metadata (dict, optional) – Metadata values to filter by. A dictionary is expected with the key corresponding to the field name found in the metadata, and the value corresponding to that field’s value.
uid (uuid.UUID, optional) – Credential UID to filter by
start_date (datetime.datetime, optional) – Filter by Credentials uploaded after the given start_date
end_date (datetime.datetime, optional) – Filter by Credentials uploaded before the given end_date
and_tags (List[str], optional) – Filter Credentials that contain all the given tags
or_tags (List[str], optional) – Filter Credentials that contain at least one of the given tags
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

typing.Any

get_file(file_hash: str, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function file_mapper>) → Any

Retrieve file details by its hash.

Parameters:

file_hash (str) – File hash to retrieve
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

get_files(mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, file_hash: ~typing.Optional[str] = None, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Retrieve all files uploaded by the Partner, or filter them via their hash.

Parameters:

mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
file_hash (str, optional) – File hash to filter by
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

T

get_job(job_uid: ~uuid.UUID, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function job_mapper>) → Any

Retrieve a specific job from the Core API, via its UUID.

Parameters:

job_uid (uuid.UUID) – Job UID to obtain
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

get_job_certificates(job_uid: ~uuid.UUID, and_tags: ~typing.Optional[str] = None, or_tags: ~typing.Optional[str] = None, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Retrieve all certificates issued in a specific Job, or filter by tags.

More information on tags on the VDX Core API official documentation.

Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

job_uid (uuid.UUID) – UID of the job
and_tags (List[str], optional) – Filter Certificates issued from credentials that contain all the given tags
or_tags (List[str], optional) – Filter Certificates issued from credentials that contain at least one of the given tags
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

typing.Any

get_job_credentials(job_uid: ~uuid.UUID, and_tags: ~typing.Optional[str] = None, or_tags: ~typing.Optional[str] = None, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, **pagination: ~typing.Dict[str, ~typing.Any]) → Optional[Any]

Retrieve all credentials issued in a specific Job, or filter by tags.

More information on tags on the VDX Core API official documentation.

Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

job_uid (uuid.UUID) – UID of the job
and_tags (List[str], optional) – Filter Credentials that contain all the given tags
or_tags (List[str], optional) – Filter Credentials that contain at least one of the given tags
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

typing.Any

get_jobs(mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, *, job_status: ~typing.Optional[str] = None, uid: ~typing.Optional[~uuid.UUID] = None, start_date: ~typing.Optional[~datetime.datetime] = None, end_date: ~typing.Optional[~datetime.datetime] = None, and_tags: ~typing.Optional[str] = None, or_tags: ~typing.Optional[str] = None, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Retrieve all partner jobs, or filter by a specific status, uid, issued date (start_date < issued_date < end_date), and tags.

More information on tags on the VDX Core API official documentation. Results are paginated, and the pagination parameters should be provided as keyword arguments. For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
job_status (str, optional) – Filter jobs with the given status
uid (uuid.UUID, optional) – Job UID to filter by
start_date (datetime.datetime, optional) – Filter Jobs that were issued after the given start_date
end_date (datetime.datetime, optional) – Filter Jobs that were issued before the given end_date
and_tags (List[str], optional) – Filter Jobs that contain all the given tags
or_tags (List[str], optional) – Filter Jobs that contain at least one of the given tags
pagination (Dict[str, Any]) – Keyword arguments for pagination

Returns:

The result of the endpoint call

Return type:

typing.Any

property header: Dict[str, Any]

Creates a header structure to be used by all requests for authentication.

Returns:: The request header
Return type:: Dict[str, Any]

issue_job(engine: str, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function job_mapper>) → Any

Immediately issues the next scheduled job for the given blockchain engine, if there are scheduled credentials.

Parameters:

engine (str) – Blockchain engine to issue
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

replace_credential_tags(replace_credential_tags: Iterable[Dict[str, List[str]]]) → None

Replaces all the given credential tags with the new ones.

Example format of the parameter:

[
    {
        "credential_uid": "81fe5e08-46e4-11ec-81d3-0242ac130003",
        "tags": ["tagA","tagB","tagC"]
    },
    {
        "credential_uid": "7f026924-b3f8-4670-bc65-37fd7da8867c",
        "tags": ["tagD"]
    }
]

Parameters:: replace_credential_tags (dict) – Dictionary containing the credential UUIDs to update and the new tags to replace the old ones with

replace_job_tags(replace_job_tags: List[Dict[str, Any]]) → None

Replaces all the given job tags with the new ones.

Example format of the parameter:

[
    {
        "job_uid": "81fe5e08-46e4-11ec-81d3-0242ac130003",
        "tags": ["tagA","tagB","tagC"]
    },
    {
        "job_uid": "7f026924-b3f8-4670-bc65-37fd7da8867c",
        "tags": ["tagD"]
    }
]

Parameters:: replace_job_tags (Dict[str, Any]) – Dictionary containing the job UUIDs to update and the new tags to replace the old ones with

revoke_certificate(cert_uid: UUID) → datetime

Revokes a certificate via its UID, making it no longer valid.

Parameters:: cert_uid (uuid.UUID) – UID of the certificate to be revoked
Returns:: The date the certificate was revoked at
Return type:: datetime.datetime

revoke_certificate_by_credential(cred_uid: UUID, engine: str) → datetime

Revokes a certificate via the credentials UID and for a specific engine, making it no longer valid.

Parameters:

cred_uid (uuid.UUID) – UID of the credential to be revoked
engine (str) – Blockchain engine to revoke the credential on

Returns:

The date the certificate was revoked at

Return type:

datetime.datetime

schedule_credentials(engine: str, credentials: ~typing.List[~uuid.UUID], mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function job_mapper>) → Any

Schedule the given list of credentials to be issued on the Blockchain engine.

Parameters:

engine (str) – Blockchain Engine to schedule the credentials
credentials (List[uuid.UUID]) – List of UUIDs of the credentials to schedule
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

update_credential_tags(updated_credential_tags: Iterable[Dict[str, List[str]]]) → None

Add new tags to one or more credentials. Previous credential tags are not changed.

Example format of the parameter:

[
    {
        "credential_uid": "81fe5e08-46e4-11ec-81d3-0242ac130003",
        "tags": ["tagA","tagB","tagC"]
    },
    {
        "credential_uid": "7f026924-b3f8-4670-bc65-37fd7da8867c",
        "tags": ["tagD"]
    }
]

Parameters:: updated_credential_tags (Dict[str, List[str]]) – Dictionary containing the credential UUIDs to update and the new tags to add

update_job_tags(updated_job_tags: List[Dict[str, Any]]) → None

Add new tags to one or more jobs. Previous job tags are not changed.

Example format of the parameter:

[
    {
        "job_uid": "81fe5e08-46e4-11ec-81d3-0242ac130003",
        "tags": ["tagA","tagB","tagC"]
    },
    {
        "job_uid": "7f026924-b3f8-4670-bc65-37fd7da8867c",
        "tags": ["tagD"]
    }
]

Parameters:: updated_job_tags (Dict[str, Any]) – Dictionary containing the job UUIDs to update and the new tags to add

upload_file(file_stream: ~typing.BinaryIO, ignore_duplicated: bool = False, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function file_mapper>) → Any

Upload a file to the Core API servers.

Parameters:

file_stream (BinaryIO) – File to be uploaded
ignore_duplicated (bool) – Flag indicating if duplicate files should be ignored. If true, and the file has already been uploaded previously, it will be re-uploaded
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

T

verify_by_certificate(filename: str, file_stream: ~typing.BinaryIO, file_content_type: str, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function verification_mapper>) → Any

Verify a certificate via its proof file.

Parameters:

filename (str) – Name of the file
file_stream (BinaryIO) – Certificate proof file stream
file_content_type (str) – Content type of the file
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any

verify_by_credential_uid(cred_uid: ~uuid.UUID, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Verify certificates via their credential UID. Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

cred_uid (uuid.UUID) – UUID of the credential
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
pagination (Dict[str, Any]) – Keyword argument containing the pagination parameters

Returns:

The result of the endpoint call

Return type:

typing.Any

verify_by_file(filename: str, file_stream: ~typing.BinaryIO, file_content_type: str, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function get_paginated_mapper.<locals>.paginated_mapper>, **pagination: ~typing.Dict[str, ~typing.Any]) → Any

Verify a certificate via the credential file. Results are paginated, and the pagination parameters should be provided as keyword arguments.

For more information on the possible parameters, check the VDX Core API official documentation.

Parameters:

filename (str) – Name of the file
file_stream (BinaryIO) – Certificate file stream
file_content_type (str) – Content type of the file
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response
pagination (Dict[str, Any]) – Keyword argument containing the pagination parameters

Returns:

The result of the endpoint call

Return type:

typing.Any

verify_by_uid(cert_uid: ~uuid.UUID, mapper: ~typing.Callable[[~typing.Dict[str, ~typing.Any]], ~typing.Any] = <function verification_mapper>) → Any

Verify a certificate via its UID.

Parameters:

cert_uid (uuid.UUID) – UUID of the certificate to verify
mapper (typing.Callable) – Optional mapper to change the format of the endpoint response

Returns:

The result of the endpoint call

Return type:

typing.Any