Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

RED-4653: Removed code that is (probably) not needed for pyinfra as a module

Browse Source
This commit is contained in:
Viktor Seifert 2022-07-18 15:32:44 +02:00
parent ebbaf6f05b
commit c286ace01c
9 changed files with 78 additions and 180 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
pyinfra/storage/adapters/adapter.py
View File

@ -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

43
pyinfra/storage/adapters/azure.py
View File

@ -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))

53
pyinfra/storage/adapters/s3.py
View File

@ -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,
))

0
pyinfra/storage/clients/__init__.py
View File

11
pyinfra/storage/clients/azure.py
View File

@ -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)

40
pyinfra/storage/clients/s3.py
View File

@ -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"(?P<protocol>https?)*(?:://)*(?P<address>(?:(?:(?:\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": <storage_endpoint>
"access_key": <storage_key>
"secret_key": <storage_secret>
}
"""
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,
)

49
pyinfra/storage/storage.py
View File

@ -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

26
pyinfra/storage/storages.py
View File

@ -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

2
requirements.txt
View File

@ -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