diff --git a/pyinfra/storage/adapters/adapter.py b/pyinfra/storage/adapters/adapter.py deleted file mode 100644 index 57cff28..0000000 --- a/pyinfra/storage/adapters/adapter.py +++ /dev/null @@ -1,34 +0,0 @@ -from abc import ABC, abstractmethod - - -class StorageAdapter(ABC): - def __init__(self, client): - self.__client = client - - @abstractmethod - def make_bucket(self, bucket_name): - raise NotImplementedError - - @abstractmethod - def has_bucket(self, bucket_name): - raise NotImplementedError - - @abstractmethod - def put_object(self, bucket_name, object_name, data): - raise NotImplementedError - - @abstractmethod - def get_object(self, bucket_name, object_name): - raise NotImplementedError - - @abstractmethod - def get_all_objects(self, bucket_name): - raise NotImplementedError - - @abstractmethod - def clear_bucket(self, bucket_name): - raise NotImplementedError - - @abstractmethod - def get_all_object_names(self, bucket_name): - raise NotImplementedError diff --git a/pyinfra/storage/adapters/azure.py b/pyinfra/storage/adapters/azure.py index 334dbaf..8f59ea7 100644 --- a/pyinfra/storage/adapters/azure.py +++ b/pyinfra/storage/adapters/azure.py @@ -3,30 +3,32 @@ from itertools import repeat from operator import attrgetter from azure.storage.blob import ContainerClient, BlobServiceClient +from retry import retry -from pyinfra.storage.adapters.adapter import StorageAdapter +from pyinfra.config import Config, get_config + +CONFIG = get_config() +logger = logging.getLogger(CONFIG.logging_level_root) -logger = logging.getLogger(__name__) logging.getLogger("azure").setLevel(logging.WARNING) logging.getLogger("urllib3").setLevel(logging.WARNING) -class AzureStorageAdapter(StorageAdapter): - def __init__(self, client): - super().__init__(client=client) - self.__client: BlobServiceClient = self._StorageAdapter__client +class AzureStorageAdapter(object): + def __init__(self, client: BlobServiceClient): + self._client: BlobServiceClient = client def has_bucket(self, bucket_name): - container_client = self.__client.get_container_client(bucket_name) + container_client = self._client.get_container_client(bucket_name) return container_client.exists() def make_bucket(self, bucket_name): - container_client = self.__client.get_container_client(bucket_name) - container_client if container_client.exists() else self.__client.create_container(bucket_name) + container_client = self._client.get_container_client(bucket_name) + container_client if container_client.exists() else self._client.create_container(bucket_name) def __provide_container_client(self, bucket_name) -> ContainerClient: self.make_bucket(bucket_name) - container_client = self.__client.get_container_client(bucket_name) + container_client = self._client.get_container_client(bucket_name) return container_client def put_object(self, bucket_name, object_name, data): @@ -35,15 +37,19 @@ class AzureStorageAdapter(StorageAdapter): blob_client = container_client.get_blob_client(object_name) blob_client.upload_blob(data, overwrite=True) + @retry(tries=3, delay=5, jitter=(1, 3)) def get_object(self, bucket_name, object_name): logger.debug(f"Downloading '{object_name}'...") - container_client = self.__provide_container_client(bucket_name) - blob_client = container_client.get_blob_client(object_name) - blob_data = blob_client.download_blob() - return blob_data.readall() + + try: + container_client = self.__provide_container_client(bucket_name) + blob_client = container_client.get_blob_client(object_name) + blob_data = blob_client.download_blob() + return blob_data.readall() + except Exception as err: + raise Exception("Failed getting object from azure client") from err def get_all_objects(self, bucket_name): - container_client = self.__provide_container_client(bucket_name) blobs = container_client.list_blobs() for blob in blobs: @@ -55,7 +61,7 @@ class AzureStorageAdapter(StorageAdapter): def clear_bucket(self, bucket_name): logger.debug(f"Clearing Azure container '{bucket_name}'...") - container_client = self.__client.get_container_client(bucket_name) + container_client = self._client.get_container_client(bucket_name) blobs = container_client.list_blobs() container_client.delete_blobs(*blobs) @@ -63,3 +69,8 @@ class AzureStorageAdapter(StorageAdapter): container_client = self.__provide_container_client(bucket_name) blobs = container_client.list_blobs() return zip(repeat(bucket_name), map(attrgetter("name"), blobs)) + + +def get_azure_storage(config: Config): + return AzureStorageAdapter( + BlobServiceClient.from_connection_string(conn_str=config.storage_azureconnectionstring)) diff --git a/pyinfra/storage/adapters/s3.py b/pyinfra/storage/adapters/s3.py index 06a1f69..46b5c2b 100644 --- a/pyinfra/storage/adapters/s3.py +++ b/pyinfra/storage/adapters/s3.py @@ -1,58 +1,77 @@ import io -from itertools import repeat import logging +from itertools import repeat from operator import attrgetter +from urllib.parse import urlparse from minio import Minio +from retry import retry +from validators import url -from pyinfra.exceptions import DataLoadingFailure -from pyinfra.storage.adapters.adapter import StorageAdapter +from pyinfra.config import Config, get_config -logger = logging.getLogger(__name__) +CONFIG = get_config() +logger = logging.getLogger(CONFIG.logging_level_root) -class S3StorageAdapter(StorageAdapter): - def __init__(self, client): - super().__init__(client=client) - self.__client: Minio = self._StorageAdapter__client +class S3StorageAdapter(object): + def __init__(self, client: Minio): + self._client = client def make_bucket(self, bucket_name): if not self.has_bucket(bucket_name): - self.__client.make_bucket(bucket_name) + self._client.make_bucket(bucket_name) def has_bucket(self, bucket_name): - return self.__client.bucket_exists(bucket_name) + return self._client.bucket_exists(bucket_name) def put_object(self, bucket_name, object_name, data): logger.debug(f"Uploading '{object_name}'...") data = io.BytesIO(data) - self.__client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes) + self._client.put_object(bucket_name, object_name, data, length=data.getbuffer().nbytes) + @retry(tries=3, delay=5, jitter=(1, 3)) def get_object(self, bucket_name, object_name): logger.debug(f"Downloading '{object_name}'...") response = None try: - response = self.__client.get_object(bucket_name, object_name) + response = self._client.get_object(bucket_name, object_name) return response.data except Exception as err: - raise DataLoadingFailure("Failed getting object from s3 client") from err + raise Exception("Failed getting object from s3 client") from err finally: if response: response.close() response.release_conn() def get_all_objects(self, bucket_name): - for obj in self.__client.list_objects(bucket_name, recursive=True): + for obj in self._client.list_objects(bucket_name, recursive=True): logger.debug(f"Downloading '{obj.object_name}'...") yield self.get_object(bucket_name, obj.object_name) def clear_bucket(self, bucket_name): logger.debug(f"Clearing S3 bucket '{bucket_name}'...") - objects = self.__client.list_objects(bucket_name, recursive=True) + objects = self._client.list_objects(bucket_name, recursive=True) for obj in objects: - self.__client.remove_object(bucket_name, obj.object_name) + self._client.remove_object(bucket_name, obj.object_name) def get_all_object_names(self, bucket_name): - objs = self.__client.list_objects(bucket_name, recursive=True) + objs = self._client.list_objects(bucket_name, recursive=True) return zip(repeat(bucket_name), map(attrgetter("object_name"), objs)) + + +def _parse_endpoint(endpoint): + if url(endpoint): + parsed_url = urlparse(endpoint) + return {"secure": parsed_url.scheme == "https", "endpoint": parsed_url.netloc} + + +def get_s3_storage(config: Config): + return S3StorageAdapter(Minio( + **_parse_endpoint(config.storage_endpoint), + access_key=config.storage_key, + secret_key=config.storage_secret, + # FIXME Is this still needed? Check and if yes, add it to config + # region=config.region, + )) diff --git a/pyinfra/storage/clients/__init__.py b/pyinfra/storage/clients/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/pyinfra/storage/clients/azure.py b/pyinfra/storage/clients/azure.py deleted file mode 100644 index 5c3751e..0000000 --- a/pyinfra/storage/clients/azure.py +++ /dev/null @@ -1,11 +0,0 @@ -from azure.storage.blob import BlobServiceClient - -from pyinfra.config import CONFIG - - -def get_azure_client(connection_string=None) -> BlobServiceClient: - - if not connection_string: - connection_string = CONFIG.storage.azure.connection_string - - return BlobServiceClient.from_connection_string(conn_str=connection_string) diff --git a/pyinfra/storage/clients/s3.py b/pyinfra/storage/clients/s3.py deleted file mode 100644 index f130989..0000000 --- a/pyinfra/storage/clients/s3.py +++ /dev/null @@ -1,40 +0,0 @@ -import re - -from minio import Minio - -from pyinfra.config import CONFIG -from pyinfra.exceptions import InvalidEndpoint - - -def parse_endpoint(endpoint): - # FIXME Greedy matching (.+) since we get random storage names on kubernetes (eg http://red-research-headless:9000) - # FIXME this has been broken and accepts invalid URLs - endpoint_pattern = r"(?Phttps?)*(?:://)*(?P
(?:(?:(?:\d{1,3}\.){3}\d{1,3})|.+)(?:\:\d+)?)" - - match = re.match(endpoint_pattern, endpoint) - - if not match: - raise InvalidEndpoint(f"Endpoint {endpoint} is invalid; expected {endpoint_pattern}") - - return {"secure": match.group("protocol") == "https", "endpoint": match.group("address")} - - -def get_s3_client(params=None) -> Minio: - """ - Args: - params: dict like - { - "endpoint": - "access_key": - "secret_key": - } - """ - if not params: - params = CONFIG.storage.s3 - - return Minio( - **parse_endpoint(params.endpoint), - access_key=params.access_key, - secret_key=params.secret_key, - region=params.region, - ) diff --git a/pyinfra/storage/storage.py b/pyinfra/storage/storage.py index c93ccf7..dcd777a 100644 --- a/pyinfra/storage/storage.py +++ b/pyinfra/storage/storage.py @@ -1,44 +1,21 @@ import logging -from retry import retry - -from pyinfra.config import CONFIG -from pyinfra.exceptions import DataLoadingFailure -from pyinfra.storage.adapters.adapter import StorageAdapter +from pyinfra.config import get_config, Config +from pyinfra.storage.adapters.azure import get_azure_storage +from pyinfra.storage.adapters.s3 import get_s3_storage +CONFIG = get_config() logger = logging.getLogger(__name__) -logger.setLevel(CONFIG.service.logging_level) +logger.setLevel(CONFIG.logging_level_root) -class Storage: - def __init__(self, adapter: StorageAdapter): - self.__adapter = adapter +def get_storage(config: Config): - def make_bucket(self, bucket_name): - self.__adapter.make_bucket(bucket_name) + if config.storage_backend == "s3": + storage = get_s3_storage(config) + elif config.storage_backend == "azure": + storage = get_azure_storage(config) + else: + raise Exception(f"Unknown storage backend '{config.storage_backend}'.") - def has_bucket(self, bucket_name): - return self.__adapter.has_bucket(bucket_name) - - def put_object(self, bucket_name, object_name, data): - self.__adapter.put_object(bucket_name, object_name, data) - - def get_object(self, bucket_name, object_name): - return self.__get_object(bucket_name, object_name) - - @retry(DataLoadingFailure, tries=3, delay=5, jitter=(1, 3)) - def __get_object(self, bucket_name, object_name): - try: - return self.__adapter.get_object(bucket_name, object_name) - except Exception as err: - logging.error(err) - raise DataLoadingFailure from err - - def get_all_objects(self, bucket_name): - return self.__adapter.get_all_objects(bucket_name) - - def clear_bucket(self, bucket_name): - return self.__adapter.clear_bucket(bucket_name) - - def get_all_object_names(self, bucket_name): - return self.__adapter.get_all_object_names(bucket_name) + return storage diff --git a/pyinfra/storage/storages.py b/pyinfra/storage/storages.py deleted file mode 100644 index 2e58186..0000000 --- a/pyinfra/storage/storages.py +++ /dev/null @@ -1,26 +0,0 @@ -from pyinfra.exceptions import UnknownStorageBackend -from pyinfra.storage.adapters.azure import AzureStorageAdapter -from pyinfra.storage.adapters.s3 import S3StorageAdapter -from pyinfra.storage.clients.azure import get_azure_client -from pyinfra.storage.clients.s3 import get_s3_client -from pyinfra.storage.storage import Storage - - -def get_azure_storage(config=None): - return Storage(AzureStorageAdapter(get_azure_client(config))) - - -def get_s3_storage(config=None): - return Storage(S3StorageAdapter(get_s3_client(config))) - - -def get_storage(storage_backend): - - if storage_backend == "s3": - storage = get_s3_storage() - elif storage_backend == "azure": - storage = get_azure_storage() - else: - raise UnknownStorageBackend(f"Unknown storage backend '{storage_backend}'.") - - return storage diff --git a/requirements.txt b/requirements.txt index 1220e73..e859054 100755 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ pika==1.2.0 +retry==0.9.2 minio==7.1.3 azure-core==1.22.1 azure-storage-blob==12.9.0 @@ -6,3 +7,4 @@ testcontainers==3.4.2 docker-compose==1.29.2 pytest~=7.0.1 funcy==1.17 +validators==0.18.2