From 81ba24d6f36cb84ab007db70e37633fddf73655a Mon Sep 17 00:00:00 2001 From: cclecle Date: Wed, 27 Mar 2024 02:17:08 +0000 Subject: [PATCH] improve quality --- src/dabdatasync/datasync.py | 90 +++++++++++++++++++++++++------------ 1 file changed, 61 insertions(+), 29 deletions(-) diff --git a/src/dabdatasync/datasync.py b/src/dabdatasync/datasync.py index ace3348..0b3464d 100644 --- a/src/dabdatasync/datasync.py +++ b/src/dabdatasync/datasync.py @@ -1,22 +1,25 @@ +"""Main datasync class""" + import json from abc import ABC, abstractmethod from typing import final, BinaryIO from typing import Self, Any, Set, Optional -from webdav3.client import Client as webdav3_Client from uuid import UUID from pathlib import Path import os -from pydantic import BaseModel import tarfile from tempfile import NamedTemporaryFile +from pydantic import BaseModel +from webdav3.client import Client as webdav3_Client + class DataSyncException(Exception): - pass + """generic datasync exception class""" class DataSyncException_InvalidManifest(DataSyncException): - pass + """specific datasync exception class - Dab appliance manifest not found""" def urljoin(*args): @@ -28,33 +31,43 @@ def urljoin(*args): return "/".join(map(lambda x: str(x).rstrip("/"), args)) -class I_DataSync_Compressor(ABC): +class A_DataSync_Compressor(ABC): + """abstract compressor class""" + @abstractmethod def compress(self, path_in: Path, file_out: Path): - pass + """compress method - virtual""" -class DataSync_Compressor_targz(I_DataSync_Compressor): +class DataSync_Compressor_targz(A_DataSync_Compressor): + """Concrete compressor class - .tar.gz compressor""" + def compress(self, path_in: Path, file_out): + """compress method - .tar.gz concrete""" with tarfile.open(fileobj=file_out, mode="w:gz") as tar: tar.add(path_in, arcname=os.path.basename(path_in)) -class A_DataSync_Record(BaseModel): +class A_DataSync_Record(BaseModel, ABC): + """Abstract DataSync Record class""" + name: str rec_type: str value: str + @abstractmethod + def compress(self, compressor: A_DataSync_Compressor, file_out: BinaryIO) -> None: + """compress the record - virtual""" + class DataSync_Record_Factory: + """DataSync Record Factory""" + ar_cls_DataSync_Record: Set[A_DataSync_Record] = set() - @classmethod - def _register_C_DataSync_Record(cls, cls_DataSync_Record): - cls.ar_cls_DataSync_Record.add(cls_DataSync_Record) - @classmethod def get_C_DataSync_Record(cls, name: str, rec_type: str, value: str) -> A_DataSync_Record | None: + """get a concrete DataSync Record class instance""" for cls_DataSync_Record in cls.ar_cls_DataSync_Record: if cls_DataSync_Record.model_fields["rec_type"].default == rec_type: return C_DataSync_Record_FS(name=name, rec_type=rec_type, value=value) @@ -62,42 +75,50 @@ class DataSync_Record_Factory: @classmethod def register(cls, _cls): - cls._register_C_DataSync_Record(_cls) + """decorator to register a concrete DataSync Record class""" + cls.ar_cls_DataSync_Record.add(_cls) return _cls @DataSync_Record_Factory.register class C_DataSync_Record_FS(A_DataSync_Record): + """Concrete DataSync Record class - FileSystem""" + rec_type: str = "fs" path: Optional[Path] = None def model_post_init(self, __context) -> None: self.path = Path(self.value) - def compress(self, compressor: I_DataSync_Compressor, file_out: BinaryIO) -> None: + def compress(self, compressor: A_DataSync_Compressor, file_out: BinaryIO) -> None: + """compress the DataSync Record - concrete FS implementation""" compressor.compress(self.path, self.path, file_out) class I_DataSync(ABC): + """Abstract DataSync class""" + manifest_path: str = "/opt/pyDABFactoryAppliance/Manifest.json" - cls_compressor: type(I_DataSync_Compressor) = DataSync_Compressor_targz + cls_compressor: type[A_DataSync_Compressor] = DataSync_Compressor_targz @classmethod @final def get_manifest_data(cls) -> dict[Any, Any]: - with open(cls.manifest_path) as f_DAB_manifest: + """utilitary method to get manifest""" + with open(cls.manifest_path, encoding="utf-8") as f_DAB_manifest: return json.load(f_DAB_manifest) @classmethod @final def try_get_instance(cls, manifest) -> Self | None: + """try to get an instance of a concrete class""" if cls.test_applicable(manifest): return cls(manifest) return None def __init__(self, manifest: dict[Any, Any]) -> None: self.connected: bool = False - self.compressor: I_DataSync_Compressor = type(self).cls_compressor() + self.compressor: A_DataSync_Compressor = type(self).cls_compressor() self.manifest: dict[Any, Any] = manifest self.app_id: UUID = UUID(manifest["APP_ID"]) self.ar_datasync_record: list[A_DataSync_Record] = [] @@ -113,33 +134,39 @@ class I_DataSync(ABC): @classmethod def test_applicable(cls, manifest: dict[Any, Any]) -> bool: + """check if a concrete class is applicable""" + del manifest # quality warning removal return False def configure(self) -> None: + """configure the class instance""" self._impl_configure() @abstractmethod def _impl_configure(self) -> None: - pass + """configure the class instance - virtual""" def connect(self) -> None: + """connect to the service""" if not self.connected: self._impl_connect() self.connected = True @abstractmethod def _impl_connect(self) -> None: - pass + """connect to the service - virtual""" def read_data(self) -> None: + """read data from the service""" self.connect() self._impl_read_data() @abstractmethod def _impl_read_data(self) -> None: - pass + """read data from the service - virtual""" def write_data(self) -> None: + """write data to the service""" self.connect() for datasync_record in self.ar_datasync_record: try: @@ -154,18 +181,17 @@ class I_DataSync(ABC): @abstractmethod def _impl_write_data(self, record_name: str, file_in: BinaryIO) -> None: - pass + """write data to the service - virtual""" class DataSync_Factory: + """DataSync Factory""" + ar_cls_DataSync: Set[I_DataSync] = set() - @classmethod - def _register_C_DataSync(cls, cls_DataSync): - cls.ar_cls_DataSync.add(cls_DataSync) - @classmethod def get_DataSync(cls) -> I_DataSync | None: + """get and configure a DataSync Concrete class instance""" manifest = I_DataSync.get_manifest_data() for cls_DataSync in cls.ar_cls_DataSync: if res := cls_DataSync.try_get_instance(manifest): @@ -175,12 +201,15 @@ class DataSync_Factory: @classmethod def register(cls, _cls): - cls._register_C_DataSync(_cls) + """decorator to register a concrete class to the factory""" + cls.ar_cls_DataSync.add(_cls) return _cls @DataSync_Factory.register class C_DataSync_NextCloud(I_DataSync): + """Concrete DataSync class - Nextcloud""" + def __init__(self, manifest: dict[Any, Any]) -> None: super().__init__(manifest) self.nextcloud_address: str @@ -197,11 +226,10 @@ class C_DataSync_NextCloud(I_DataSync): if manifest["Args"]["FSSync_NextCloud_Enabled"]["value"] is True: return True return False - else: - raise DataSyncException_InvalidManifest() - return True + raise DataSyncException_InvalidManifest() def _impl_configure(self): + """configure the class instance - concrete implementation""" if "FSSync_NextCloud_Address" in self.manifest["Args"]: self.nextcloud_address = self.manifest["Args"]["FSSync_NextCloud_Address"]["value"] else: @@ -223,12 +251,14 @@ class C_DataSync_NextCloud(I_DataSync): raise DataSyncException_InvalidManifest() def _impl_connect(self): + """connect to the remote service - concrete implementation""" full_adress = urljoin(self.nextcloud_address, "remote.php/dav/files/", self.nextcloud_user) self.client = webdav3_Client( {"webdav_hostname": full_adress, "webdav_login": self.nextcloud_user, "webdav_password": self.nextcloud_password} ) def _check_create_dir(self): + """check and create directory in remote service""" url_accumulator: str = "" for url_part in self.nextcloud_path.split("/"): url_accumulator += "/" + url_part @@ -236,9 +266,11 @@ class C_DataSync_NextCloud(I_DataSync): self.client.mkdir(url_accumulator) def _impl_read_data(self): + """read data from the remote service - concrete implementation""" self._check_create_dir() def _impl_write_data(self, record_name: str, file_in: BinaryIO): + """write data to the remote service - concrete implementation""" self._check_create_dir() self.client.upload_sync( remote_path=str(Path(self.nextcloud_path) / record_name).replace(os.sep, "/"),