removing old storage logic WIP
This commit is contained in:
Julius Unverfehrt
parent
880e22832c
commit
c530b5929a
@ -46,7 +46,7 @@ class S3StorageAdapter(StorageAdapter):
|
||||
|
||||
def clear_bucket(self, bucket_name):
|
||||
logger.debug(f"Clearing S3 bucket '{bucket_name}'...")
|
||||
objects = self.__client.list_objects(bucket_name)
|
||||
objects = self.__client.list_objects(bucket_name, recursive=True)
|
||||
for obj in objects:
|
||||
self.__client.remove_object(bucket_name, obj.object_name)
|
||||
|
||||
|
||||
@ -1,121 +0,0 @@
|
||||
"""Implements a wrapper around an Azure blob storage client that provides operations on the blob storage required by the service."""
|
||||
import os
|
||||
from typing import Iterable
|
||||
|
||||
from azure.storage.blob import BlobServiceClient, ContainerClient
|
||||
|
||||
from pyinfra.config import CONFIG
|
||||
from pyinfra.storage.storage import StorageHandle
|
||||
|
||||
|
||||
def get_blob_service_client(connection_string=None) -> BlobServiceClient:
|
||||
"""Instantiates a minio.Minio client.
|
||||
|
||||
Args:
|
||||
connection_string: Azure blob storage connection string.
|
||||
|
||||
Returns:
|
||||
A minio.Minio client.
|
||||
"""
|
||||
connection_string = CONFIG.azure_blob_storage.connection_string if connection_string is None else connection_string
|
||||
return BlobServiceClient.from_connection_string(conn_str=connection_string)
|
||||
|
||||
|
||||
class AzureBlobStorageHandle(StorageHandle):
|
||||
"""Implements a wrapper around an Azure blob storage client that provides operations on the blob storage required by
|
||||
the service.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initializes an AzureBlobStorageHandle"""
|
||||
super().__init__()
|
||||
self.client: BlobServiceClient = get_blob_service_client()
|
||||
self.default_container_name = CONFIG.azure_blob_storage.container
|
||||
self.backend = "azure"
|
||||
|
||||
def __get_container_client(self, container_name) -> ContainerClient:
|
||||
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
container_client = self._StorageHandle__provide_container(container_name)
|
||||
return container_client
|
||||
|
||||
def _StorageHandle__provide_container(self, container_name: str):
|
||||
container_client = self.client.get_container_client(container_name)
|
||||
return container_client if container_client.exists() else self.client.create_container(container_name)
|
||||
|
||||
def _StorageHandle__add_file(self, path, storage_path, container_name: str = None):
|
||||
|
||||
container_client = self.__get_container_client(container_name)
|
||||
blob_client = container_client.get_blob_client(storage_path)
|
||||
|
||||
with open(path, "rb") as f:
|
||||
blob_client.upload_blob(f, overwrite=True)
|
||||
|
||||
def list_files(self, container_name=None) -> Iterable[str]:
|
||||
"""List all files in a container.
|
||||
|
||||
Args:
|
||||
container_name: container to list files from.
|
||||
|
||||
Returns:
|
||||
Iterable of filenames.
|
||||
"""
|
||||
return self._list_files("name", container_name=container_name)
|
||||
|
||||
def get_objects(self, container_name=None):
|
||||
"""List all files in a container_name.
|
||||
|
||||
Args:
|
||||
container_name: Container to list files from.
|
||||
|
||||
Returns:
|
||||
Iterable over all objects in the container.
|
||||
"""
|
||||
container_client = self.__get_container_client(container_name)
|
||||
blobs = container_client.list_blobs()
|
||||
yield from blobs
|
||||
|
||||
def _StorageHandle__list_containers(self):
|
||||
return self.client.list_containers()
|
||||
|
||||
def _StorageHandle__purge(self) -> None:
|
||||
"""Deletes all files and buckets in the store."""
|
||||
for container in self.client.list_containers():
|
||||
self.client.delete_container(container.name)
|
||||
|
||||
def _StorageHandle__fget_object(self, container_name: str, object_name: str, target_path):
|
||||
|
||||
with open(target_path, "wb") as f:
|
||||
blob_data = self.get_object(container_name, object_name)
|
||||
blob_data.readinto(f)
|
||||
|
||||
def get_object(self, object_name: str, container_name: str = None):
|
||||
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
container_client = self.__get_container_client(container_name)
|
||||
blob_client = container_client.get_blob_client(object_name)
|
||||
blob_data = blob_client.download_blob()
|
||||
|
||||
return blob_data
|
||||
|
||||
def _StorageHandle__remove_file(self, folder: str, filename: str, container_name: str = None) -> None:
|
||||
"""Removes a file from the store.
|
||||
|
||||
Args:
|
||||
folder: Folder containing file.
|
||||
filename: Name of file (without folder) to remove.
|
||||
container_name: container containing file.
|
||||
"""
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
path = os.path.join(folder, filename)
|
||||
|
||||
container_client = self.__get_container_client(container_name)
|
||||
blob_client = container_client.get_blob_client(path)
|
||||
if blob_client.exists():
|
||||
container_client.delete_blob(blob_client)
|
||||
@ -1,119 +0,0 @@
|
||||
"""Implements a wrapper around a MinIO client that provides operations on the MinIO store required by the service."""
|
||||
import os
|
||||
from typing import Iterable
|
||||
|
||||
from minio import Minio
|
||||
|
||||
from pyinfra.config import CONFIG
|
||||
from pyinfra.storage.storage import StorageHandle
|
||||
|
||||
|
||||
def get_minio_client(access_key=None, secret_key=None) -> Minio:
|
||||
"""Instantiates a minio.Minio client.
|
||||
|
||||
Args:
|
||||
access_key: Access key for MinIO client (username).
|
||||
secret_key: Secret key for MinIO client (password).
|
||||
|
||||
Returns:
|
||||
A minio.Minio client.
|
||||
"""
|
||||
access_key = CONFIG.minio.user if access_key is None else access_key
|
||||
secret_key = CONFIG.minio.password if secret_key is None else secret_key
|
||||
|
||||
# TODO: secure=True/False?
|
||||
return Minio(CONFIG.minio.endpoint, access_key=access_key, secret_key=secret_key, secure=False)
|
||||
|
||||
|
||||
class MinioHandle(StorageHandle):
|
||||
"""Wrapper around a MinIO client that provides operations on the MinIO store required by the service."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initializes a MinioHandle"""
|
||||
super().__init__()
|
||||
self.client: Minio = get_minio_client()
|
||||
self.default_container_name = CONFIG.minio.bucket
|
||||
self.backend = "s3"
|
||||
|
||||
def _StorageHandle__provide_container(self, container_name):
|
||||
if not self.client.bucket_exists(container_name):
|
||||
self.client.make_bucket(container_name)
|
||||
|
||||
def _StorageHandle__add_file(self, path, storage_path, container_name=None):
|
||||
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
self._StorageHandle__provide_container(container_name)
|
||||
|
||||
with open(path, "rb") as f:
|
||||
stat = os.stat(path)
|
||||
self.client.put_object(container_name, storage_path, f, stat.st_size)
|
||||
|
||||
def list_files(self, container_name=None) -> Iterable[str]:
|
||||
"""List all files in a container.
|
||||
|
||||
Args:
|
||||
container_name: container to list files from.
|
||||
|
||||
Returns:
|
||||
Iterable of filenames.
|
||||
"""
|
||||
return self._list_files("object_name", container_name=container_name)
|
||||
|
||||
def get_objects(self, container_name=None):
|
||||
"""Gets all objects in a container.
|
||||
|
||||
Args:
|
||||
container_name: container to get objects from.
|
||||
|
||||
Returns:
|
||||
Iterable over all objects in the container.
|
||||
"""
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
yield from self.client.list_objects(container_name, recursive=True)
|
||||
|
||||
def _StorageHandle__list_containers(self):
|
||||
return self.client.list_buckets()
|
||||
|
||||
def _StorageHandle__purge(self) -> None:
|
||||
"""Deletes all files and containers in the store."""
|
||||
for container, obj in self.get_all_objects():
|
||||
self.client.remove_object(container.name, obj.object_name)
|
||||
|
||||
for container in self.client.list_buckets():
|
||||
self.client.remove_bucket(container.name)
|
||||
|
||||
def _StorageHandle__fget_object(self, container_name, object_name, target_path):
|
||||
self.client.fget_object(container_name, object_name, target_path)
|
||||
|
||||
def get_object(self, object_name, container_name=None):
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
response = None
|
||||
|
||||
try:
|
||||
response = self.client.get_object(container_name, object_name)
|
||||
return response.data
|
||||
finally:
|
||||
if response:
|
||||
response.close()
|
||||
response.release_conn()
|
||||
|
||||
def _StorageHandle__remove_file(self, folder: str, filename: str, container_name: str = None) -> None:
|
||||
"""Removes a file from the store.
|
||||
|
||||
Args:
|
||||
folder: Folder containing file.
|
||||
filename: Name of file (without folder) to remove.
|
||||
container_name: container containing file.
|
||||
"""
|
||||
if container_name is None:
|
||||
container_name = self.default_container_name
|
||||
|
||||
path = os.path.join(folder, filename)
|
||||
|
||||
if self.client.bucket_exists(container_name):
|
||||
self.client.remove_object(container_name, path)
|
||||
@ -22,188 +22,3 @@ class Storage:
|
||||
|
||||
def list_bucket_files(self, bucket_name):
|
||||
return self.__adapter.list_bucket_files(bucket_name)
|
||||
|
||||
|
||||
# import abc
|
||||
# import gzip
|
||||
# import logging
|
||||
# import os
|
||||
# from itertools import repeat
|
||||
# from operator import attrgetter
|
||||
# from typing import Iterable
|
||||
#
|
||||
# from pyinfra.utils.file import path_to_compressed_storage_pdf_object_name, provide_directory
|
||||
# from pyinfra.utils.retry import NoAttemptsLeft, max_attempts
|
||||
#
|
||||
#
|
||||
# class StorageHandle:
|
||||
# """Storage API base"""
|
||||
#
|
||||
# def __init__(self):
|
||||
# self.default_container_name = None
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __provide_container(self, container_name):
|
||||
# pass
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __add_file(self, path, filename, container_name=None):
|
||||
# pass
|
||||
#
|
||||
# @max_attempts(n_attempts=10, max_timeout=60)
|
||||
# def add_file(self, path: str, folder: str = None, container_name: str = None) -> None:
|
||||
# """Adds a file to the store.
|
||||
#
|
||||
# Args:
|
||||
# path: Path to file to add to store.
|
||||
# folder: Folder to hold file.
|
||||
# container_name: container to hold file.
|
||||
# """
|
||||
# storage_path = self.__storage_path(path, folder=folder)
|
||||
# self.__add_file(path, storage_path, container_name)
|
||||
#
|
||||
# @max_attempts()
|
||||
# def _list_files(self, object_name_attr="object_name", container_name=None) -> Iterable[str]:
|
||||
# """List all files in a container.
|
||||
#
|
||||
# Args:
|
||||
# container_name: container to list files from.
|
||||
#
|
||||
# Returns:
|
||||
# Iterable of filenames.
|
||||
# """
|
||||
# if container_name is None:
|
||||
# container_name = self.default_container_name
|
||||
# return map(attrgetter(object_name_attr), self.get_objects(container_name))
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def list_files(self, container_name=None) -> Iterable[str]:
|
||||
# pass
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def get_objects(self, container_name=None):
|
||||
# pass
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __list_containers(self):
|
||||
# pass
|
||||
#
|
||||
# @max_attempts()
|
||||
# def get_all_objects(self) -> Iterable:
|
||||
# """Gets all objects in the store
|
||||
#
|
||||
# Returns:
|
||||
# Iterable over all objects in the store.
|
||||
# """
|
||||
# for container in self.__list_containers():
|
||||
# yield from zip(repeat(container), self.get_objects(container.name))
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __purge(self) -> None:
|
||||
# pass
|
||||
#
|
||||
# @max_attempts()
|
||||
# def purge(self) -> None:
|
||||
# self.__purge()
|
||||
#
|
||||
# def list_files_by_type(self, container_name=None, extension=".pdf.gz"):
|
||||
# return filter(lambda p: p.endswith(extension), self.list_files(container_name))
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __fget_object(self, *args, **kwargs):
|
||||
# pass
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def get_object(self, *args, **kwargs):
|
||||
# pass
|
||||
#
|
||||
# @staticmethod
|
||||
# def __storage_path(path, folder: str = None):
|
||||
# def path_to_filename(path):
|
||||
# return os.path.basename(path)
|
||||
#
|
||||
# storage_path = path_to_filename(path)
|
||||
# if folder is not None:
|
||||
# storage_path = os.path.join(folder, storage_path)
|
||||
#
|
||||
# return storage_path
|
||||
#
|
||||
# @max_attempts()
|
||||
# def list_folders_and_files(self, container_name: str = None) -> Iterable[str]:
|
||||
# """Lists pairs of folder name (dossier-IDs) and file name (file-IDs) of items in a container.
|
||||
#
|
||||
# Args:
|
||||
# container_name: container to list items for.
|
||||
#
|
||||
# Returns:
|
||||
# Iterable of pairs folder name (dossier-ID) and file names (file-ID)
|
||||
# """
|
||||
# return map(lambda p: p.split("/"), self.list_files_by_type(container_name))
|
||||
#
|
||||
# @abc.abstractmethod
|
||||
# def __remove_file(self, folder: str, filename: str, container_name: str = None) -> None:
|
||||
# pass
|
||||
#
|
||||
# @max_attempts()
|
||||
# def remove_file(self, folder: str, filename: str, container_name: str = None) -> None:
|
||||
# self.__remove_file(folder, filename, container_name)
|
||||
#
|
||||
# def add_file_compressed(self, path, folder: str = None, container_name: str = None) -> None:
|
||||
# """Adds a file as a .gz archive to the store.
|
||||
#
|
||||
# Args:
|
||||
# path: Path to file to add to store.
|
||||
# folder: Folder to hold file.
|
||||
# container_name: container to hold file.
|
||||
# """
|
||||
#
|
||||
# def compress(path_in: str, path_out: str):
|
||||
# with open(path_in, "rb") as f_in, gzip.open(path_out, "wb") as f_out:
|
||||
# f_out.writelines(f_in)
|
||||
#
|
||||
# path_gz = path_to_compressed_storage_pdf_object_name(path)
|
||||
# compress(path, path_gz)
|
||||
#
|
||||
# self.add_file(path_gz, folder, container_name)
|
||||
# os.unlink(path_gz)
|
||||
#
|
||||
# @max_attempts()
|
||||
# def download_file(self, object_names: str, target_root_dir: str, container_name: str = None) -> str:
|
||||
# """Downloads a file from the store.
|
||||
#
|
||||
# Args:
|
||||
# object_names: Complete object name (folder and file).
|
||||
# target_root_dir: Root directory to download file into (including its folder).
|
||||
# container_name: container to load file from.
|
||||
#
|
||||
# Returns:
|
||||
# str: Path to downloaded file.
|
||||
# """
|
||||
#
|
||||
# @max_attempts(5, exceptions=(FileNotFoundError,))
|
||||
# def download(object_name: str) -> str:
|
||||
#
|
||||
# path, basename = os.path.split(object_name)
|
||||
# target_dir = os.path.join(target_root_dir, path)
|
||||
# provide_directory(target_dir)
|
||||
# target_path = os.path.join(target_dir, basename)
|
||||
#
|
||||
# logging.log(msg=f"Downloading {object_name}...", level=logging.DEBUG)
|
||||
# try:
|
||||
# self.__fget_object(container_name, object_name, target_path)
|
||||
# logging.log(msg=f"Downloaded {object_name}.", level=logging.DEBUG)
|
||||
# except Exception as err:
|
||||
# logging.log(msg=f"Downloading {object_name} failed.", level=logging.ERROR)
|
||||
# raise err
|
||||
# return target_path
|
||||
#
|
||||
# if container_name is None:
|
||||
# container_name = self.default_container_name
|
||||
#
|
||||
# try:
|
||||
# target_path = download(object_names)
|
||||
# except NoAttemptsLeft as err:
|
||||
# logging.log(msg=f"{err}", level=logging.ERROR)
|
||||
# raise err
|
||||
#
|
||||
# return target_path
|
||||
|
||||
@ -14,6 +14,7 @@ from pyinfra.test.storage.adapter_mock import StorageAdapterMock
|
||||
from pyinfra.test.storage.client_mock import StorageClientMock
|
||||
|
||||
|
||||
#TODO Add test for recursive clearing & getting
|
||||
@pytest.mark.parametrize("client_name", ["mock", "azure", "s3"])
|
||||
class TestStorage:
|
||||
def test_clearing_bucket_yields_empty_bucket(self, storage, bucket_name):
|
||||
|
||||
@ -3,16 +3,6 @@
|
||||
import os
|
||||
|
||||
|
||||
def provide_directory(path):
|
||||
if os.path.isfile(path):
|
||||
provide_directory(os.path.dirname(path))
|
||||
if not os.path.isdir(path):
|
||||
try:
|
||||
os.makedirs(path)
|
||||
except FileExistsError:
|
||||
pass
|
||||
|
||||
|
||||
def produce_compressed_storage_pdf_object_name(path_no_ext, ext="pdf"):
|
||||
return f"{path_no_ext}.ORIGIN.{ext}.gz"
|
||||
|
||||
@ -21,9 +11,3 @@ def dossier_id_and_file_id_to_compressed_storage_pdf_object_name(dossier_id, fil
|
||||
path_no_ext = os.path.join(dossier_id, file_id)
|
||||
pdf_object_name = produce_compressed_storage_pdf_object_name(path_no_ext)
|
||||
return pdf_object_name
|
||||
|
||||
|
||||
def path_to_compressed_storage_pdf_object_name(path):
|
||||
path_no_ext, ext = os.path.splitext(path)
|
||||
path_gz = produce_compressed_storage_pdf_object_name(path_no_ext)
|
||||
return path_gz
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user