ae.cloud_storage

distribute files to and retrieve them from cloud storage hosts.

supported cloud storage hosts: * Google Drive: https://developers.google.com/drive/api/guides/about-sdk * DigiStorage: https://storage.rcs-rds.ro/help/developers

a comparison of cloud storage hosts, including free ones, can be found here: https://comparisontabl.es/cloud-storage/ (or https://docs.google.com/spreadsheets/d/1cEd65XDW3gBHnRsJ0rbq3V_B28mKySHiMPAZvArHiiA/). but not all of them offer an API, see some recommendations in: * https://blog.apilayer.com/5-cloud-storage-apis/#Filestack_API * https://www.jsonapi.co/public-api/category/Cloud%20Storage%20&%20File%20Sharing

Functions

csh_api_class(csh_id)

determine from the specified cloud storage host id the associated api class

Classes

CshApiBase(**csh_args)

abstract Cloud storage host api base class.

DigiApi([root_folder, email, password])

upload, update, download and delete files from Digi Movil Storage.

GoodriveApi([root_folder, sa_cred_dict, ...])

upload, update, download and delete files from a Google Drive.

_registered_csh_classes: dict[str, type[CshApiBase]] = {'Digi': <class 'ae.cloud_storage.DigiApi'>, 'Goodrive': <class 'ae.cloud_storage.GoodriveApi'>}

mapping of cloud-storage-hosts class ids to their related api classes

class CshApiBase(**csh_args)[source]

Bases: ErrorMsgMixin, ABC

abstract Cloud storage host api base class.

__init__(**csh_args)[source]

cloud storage host api instantiation and argument check.

Parameters:

csh_args – individual arguments, like host root path and credentials, of a cloud storage host api.

classmethod __init_subclass__()[source]

This method is called when a class is subclassed.

The default implementation does nothing. It may be overridden to extend subclasses.

abstractmethod deployed_file_content(file_path)[source]

determine the file content of a file deployed to a server.

Parameters:

file_path (str) – path of a deployed file relative to the host root.

Return type:

Optional[bytes]

Returns:

file content as bytes or None if error occurred (check self.error_message).

abstractmethod deploy_file(file_path, source_path='')[source]

add or update a binary file to the cloud storage host.

Parameters:
  • file_path (str) – path (relative to the host root) and name of the file to be deployed (added or updated).

  • source_path (str) – source path if differs from the destination path given in file_path.

Return type:

str

Returns:

created/updated file path or empty string on error (see self.error_message for details).

abstractmethod delete_file_or_folder(file_path)[source]

delete a file or folder on the cloud storage host.

Parameters:

file_path (str) – path relative to the host root of the file/folder to be deleted. to delete a folder a trailing slash character has to be added. .. note:: deleting a folder will also delete all files/folders underneath it.

Return type:

str

Returns:

error message if deletion failed else on success an empty string.

csh_api_class(csh_id)[source]

determine from the specified cloud storage host id the associated api class

Parameters:

csh_id (str) – id of the cloud storage api class.

Return type:

Type[CshApiBase]

Returns:

cloud storage api class if registered/declared, else the abstract class CshApiClass.

class DigiApi(root_folder='', email='', password='', **csh_args)[source]

Bases: CshApiBase

upload, update, download and delete files from Digi Movil Storage.

the requests package is the only requirement of this class, which can be installed via pip install requests.

__init__(root_folder='', email='', password='', **csh_args)[source]

DigiStorage host instantiation.

Parameters:
  • root_folder (str) – host root folder name/path.

  • email (str) – host account email address to authenticate.

  • password (str) – host account password to authenticate.

  • csh_args – generic cloud storage host arguments.

_create_dirs(folder_path)[source]
Return type:

str

_request(method, slug, path, **kwargs)[source]
Return type:

Optional[Response]

delete_file_or_folder(file_path)[source]

delete a file or folder on the cloud storage host.

Parameters:

file_path (str) – path relative to the host root of the file/folder to be deleted. to delete a folder a trailing slash character has to be added. to delete the root folder pass an empty string or a single slash character. .. note:: ¡deleting a folder will also delete all files/folders underneath it!

Return type:

str

Returns:

error message if deletion failed else on success an empty string.

deployed_file_content(file_path)[source]

determine the file content of a file deployed to a server.

Parameters:

file_path (str) – path of a deployed file relative to the host root path.

Return type:

Optional[bytes]

Returns:

file content as bytes or None if error occurred (check self.error_message).

deploy_file(file_path, source_path='')[source]

add or update a binary file to the cloud storage host.

Parameters:
  • file_path (str) – path (relative to the host root) and name of the file to be deployed (added or updated).

  • source_path (str) – source path if differs from the destination path given in file_path.

Return type:

str

Returns:

created/updated file path or empty string on error (see self.error_message for details).

list_dir(folder_path)[source]

determine files and folders in the specified folder.

Parameters:

folder_path (str) – path to the folder (relative to the host root folder) to determine items of.

Return type:

Optional[list[str]]

Returns:

list of files/folders names in the specified folder (sub-folders have a trailing slash) or None if the folder does not exist.

__del__()[source]
class GoodriveApi(root_folder='root', sa_cred_dict=None, sa_cred_file='.service_account_credentials.json', oa_cred_file='.oauth2_credentials.json', **csh_args)[source]

Bases: CshApiBase

upload, update, download and delete files from a Google Drive.

Note

the Google Drive root folder (id) cannot be created programmatically; prepare before run the unit tests.

to prepare Google Drive host api instance (root folder and authentication), create in your console (at https://console.cloud.google.com/apis/credentials?orgonly=true&project=oaio-project&supportedpurview=organizationId) the credentials for a service account (or an OAuth 2.0 Client) and activate it for your Google Drive project. then download the credential JSON file and store it, and specify its location in the instantiation of the class.

Hint

the Google API Keys cannot be used for Google Drive authentication.

to install the required packages run in your virtual environment the following command:

pip install google-auth-oauthlib google-auth-httplib2 google-api-python-client
CRED_SCOPES = ['https://www.googleapis.com/auth/drive']
FOLDER_MIMETYPE = 'application/vnd.google-apps.folder'
GOOGLE_DRIVE_DEFAULT_ROOT_FOLDER = 'root'
SKIPPED_FILES_MIMETYPE_PREFIX = 'application/vnd.google-apps.'
GoodriveRequestReturnType

alias of dict[str, Any]

__init__(root_folder='root', sa_cred_dict=None, sa_cred_file='.service_account_credentials.json', oa_cred_file='.oauth2_credentials.json', **csh_args)[source]

initialize an instance to access the Google Drive via API.

Parameters:
  • root_folder (str) – Google Drive root folder id for this host api instance.

  • sa_cred_dict (Optional[dict[str, Any]]) – service account credentials dictionary (if passed has priority over credential files).

  • sa_cred_file (str) – service account credentials file path (if exists has priority over oa_cred_file).

  • oa_cred_file (str) – path to the OAuth 2.0 credentials file.

_create_folder(folder_name, parent_id)[source]

create a single folder under the parent folder (specified by id).

Note

sometimes folder_file_ids() fails to see a created folder after the deployment. on subsequent bulk creations use wait_for_deployment_finish() to wait for finished/completed folder creation.

Parameters:
  • folder_name (str) – name of the folder to create.

  • parent_id (str) – id of the parent folder where to create a new folder underneath.

Return type:

dict[str, Any]

Returns:

dict with the id of the created folder.

_oauth2_authenticate(cred_info, cached_cred_file_path='.token.json')[source]

ALTERNATIVE EXPERIMENTAL OAuth2 authentication - missing unit tests

Return type:

Optional[Credentials]

property _files
_service_account_authenticate(cred_info)[source]
Return type:

Optional[Credentials]

_request(prepared_call)[source]
Return type:

dict[str, Any]

delete_file_or_folder(file_path, empty_trash=False)[source]

delete a file or folder on the cloud storage host.

Parameters:
  • file_path (str) – path relative to the host root of the file/folder to be deleted. to delete a folder a trailing slash character has to be added. .. note:: deleting a folder will also delete all files/folders underneath it.

  • empty_trash (bool) – pass True to empty the trash (definitely removing this and other deleted files).

Return type:

str

Returns:

error message if deletion failed else on success an empty string.

deployed_file_content(file_path)[source]

determine the file content of a file deployed to a server.

Parameters:

file_path (str) – path of a deployed file relative to the host root folder.

Return type:

Optional[bytes]

Returns:

file content as bytes or None if error occurred (check self.error_message).

deploy_file(file_path, source_path='')[source]

add or update a binary file to the cloud storage host.

Parameters:
  • file_path (str) – path (relative to the host root) and name of the file to be deployed (added or updated).

  • source_path (str) – source path if differs from the destination path given in file_path.

Return type:

str

Returns:

created/updated file id or empty string on error (check self.error_message for details).

Note

sometimes folder_file_ids() fails to see a deployed file directly after the deployment. on subsequent bulk deployments use wait_for_deployment_finish() to wait for finished/completed deployment.

folder_file_ids(file_path, folder_id='', create_folders=False)[source]

convert Google Drive file path into the related folder and file ids.

Parameters:
  • file_path (str) – path of a file or folder, relative to the folder specified in folder_id.

  • folder_id (str) –

    Google Drive folder id of the root folder where folder_path is underneath of. if not passed then it defaults to the root folder (specified via __init__()).

    Note

    MyDrive GOOGLE_DRIVE_DEFAULT_ROOT_FOLDER/’root’ id is working for OAuth2, but not for service accounts authentication.

  • create_folders (bool) – pass True to create non-existing folders in the specified file path.

Return type:

tuple[str, str]

Returns:

tuple of ids of the specified folder and of the basename file/folder. if the second tuple item is an empty string then an error occurred, because either the specified folder/file does not exist, or the specified basename ends with a slash, although it is a file.

Note

if folder_path specifies a Google Drive cloud mimetype, like Google Doc/Sheet/.., then although the file id get returned, an error message get set (stating that Google Docs cannot be downloaded as files via deployed_file_content()).

wait_for_deployment_finish(path, file_id='', max_tries=99, verbose=False)[source]

wait until the deployment of a file or the creation of a folder is fully visible in the api.

Parameters:
  • path (str) – path to the just created/deployed file/folder.

  • file_id (str) – pass file/folder id to wait for, to restrict the item specified in path.

  • max_tries (int) – number of tries/loops to check if the api can see the file/folder.

  • verbose (bool) – pass True to print to the console for each try.

Return type:

tuple[tuple[str, str], int, float]

Returns:

3-tuple consisting of the return value of folder_file_ids(), the number of tries and the total amount of seconds waited.