Source code for dtlpy.entities.driver

import logging
import attr
from enum import Enum
from collections import namedtuple
from .. import entities, repositories
from ..services.api_client import ApiClient

logger = logging.getLogger(name='dtlpy')


[docs]class ExternalStorage(str, Enum): """ The type of the Integration. .. list-table:: :widths: 15 150 :header-rows: 1 * - State - Description * - S3 - AWS S3 drivers * - GCS - Google GCS drivers * - AZUREBLOB - Microsoft AZURE BLOB drivers * - AZURE_DATALAKE_GEN2 - Microsoft AZURE GEN2 drivers """ S3 = "s3" GCS = "gcs" AZUREBLOB = "azureblob" AZURE_DATALAKE_GEN2 = 'azureDatalakeGen2' KEY_VALUE = "key_value" AWS_STS = 'aws-sts'
[docs]@attr.s() class Driver(entities.BaseEntity): """ Driver entity """ creator = attr.ib() allow_external_delete = attr.ib() allow_external_modification = attr.ib() created_at = attr.ib() type = attr.ib() integration_id = attr.ib() integration_type = attr.ib() metadata = attr.ib(repr=False) name = attr.ib() id = attr.ib() path = attr.ib() # api _client_api = attr.ib(type=ApiClient, repr=False) _repositories = attr.ib(repr=False) @_repositories.default def set_repositories(self): reps = namedtuple('repositories', field_names=['drivers']) return reps( drivers=repositories.Drivers(client_api=self._client_api), ) @property def drivers(self): assert isinstance(self._repositories.drivers, repositories.Drivers) return self._repositories.drivers
[docs] @classmethod def from_json(cls, _json, client_api, is_fetched=True): """ Build a Driver entity object from a json :param _json: _json response from host :param client_api: ApiClient entity :param is_fetched: is Entity fetched from Platform :return: Driver object """ inst = cls(creator=_json.get('creator', None), allow_external_delete=_json.get('allowExternalDelete', None), allow_external_modification=_json.get('allowExternalModification', None), created_at=_json.get('createdAt', None), type=_json.get('type', None), integration_id=_json.get('integrationId', None), integration_type=_json.get('integrationType', None), metadata=_json.get('metadata', None), name=_json.get('name', None), id=_json.get('id', None), client_api=client_api, path=_json.get('path', None)) inst.is_fetched = is_fetched return inst
[docs] def to_json(self): """ Returns platform _json format of object :return: platform json format of object :rtype: dict """ output_dict = attr.asdict(self, filter=attr.filters.exclude(attr.fields(Driver)._client_api, attr.fields(Driver).allow_external_delete, attr.fields(Driver).allow_external_modification, attr.fields(Driver).created_at, attr.fields(Driver).integration_id, attr.fields(Driver).integration_type, attr.fields(Driver).path )) output_dict['allowExternalDelete'] = self.allow_external_delete output_dict['allowExternalModification'] = self.allow_external_modification output_dict['createdAt'] = self.created_at output_dict['integrationId'] = self.integration_id output_dict['integrationType'] = self.integration_type if self.path is not None: output_dict['path'] = self.path return output_dict
[docs] def delete(self, sure=False, really=False): """ Delete a driver forever! **Prerequisites**: You must be an *owner* or *developer* to use this method. :param bool sure: are you sure you want to delete? :param bool really: really really? :return: True if success :rtype: bool **Example**: .. code-block:: python driver.delete(sure=True, really=True) """ return self.drivers.delete(driver_id=self.id, sure=sure, really=really)
[docs]@attr.s() class AzureBlobDriver(Driver): container_name = attr.ib(default=None)
[docs] def to_json(self): _json = super().to_json() _json['containerName'] = self.container_name return _json
[docs] @classmethod def from_json(cls, _json, client_api, is_fetched=True): """ Build a Driver entity object from a json :param _json: _json response from host :param client_api: ApiClient entity :param is_fetched: is Entity fetched from Platform :return: Driver object """ inst = super().from_json(_json, client_api, is_fetched=True) inst.container_name = _json.get('containerName', None) return inst
[docs]@attr.s() class GcsDriver(Driver): bucket = attr.ib(default=None)
[docs] def to_json(self): _json = super().to_json() _json['bucket'] = self.bucket return _json
[docs] @classmethod def from_json(cls, _json, client_api, is_fetched=True): """ Build a Driver entity object from a json :param _json: _json response from host :param client_api: ApiClient entity :param is_fetched: is Entity fetched from Platform :return: Driver object """ inst = super().from_json(_json, client_api, is_fetched=True) inst.bucket = _json.get('bucket', None) return inst
[docs]@attr.s() class S3Driver(Driver): bucket_name = attr.ib(default=None) region = attr.ib(default=None) storage_class = attr.ib(default=None)
[docs] def to_json(self): _json = super().to_json() _json['bucketName'] = self.bucket_name if self.region is not None: _json['region'] = self.region if self.storage_class is not None: _json['storageClass'] = self.storage_class return _json
[docs] @classmethod def from_json(cls, _json, client_api, is_fetched=True): """ Build a Driver entity object from a json :param _json: _json response from host :param client_api: ApiClient entity :param is_fetched: is Entity fetched from Platform :return: Driver object """ inst = super().from_json(_json, client_api, is_fetched=True) inst.bucket_name = _json.get('bucketName', None) inst.region = _json.get('region', None) inst.storage_class = _json.get('storageClass', None) return inst