I wrote some code and thought I may get better with some feedback. I program for almost 5 years, mainly in python.
I care most about the unit testing. I am not really sure, how industry standard unit tests in python look like.
storable.py
import os
import uuid
import logging
from os.path import *
from typing import Type, List, Optional, Iterable, Union
from abc import ABC, abstractmethod
log = logging.getLogger(__name__)
# TODO: remove
class DirectorySave(ABC):
def __init__(self, uuid_: bytes = None):
if uuid_ is None:
uuid_ = uuid.uuid4()
else:
uuid_ = uuid.UUID(bytes=uuid_)
self.uuid = uuid_
@abstractmethod
def save(self, path: str, force: bool = False):
"""
This method saves this object to a directory.
:param path: this should be a string, and a valid and existing path
:param force: if this is enabled, any error will not crash the saving process,
whether skip the failing object
:return: the save path
"""
@classmethod
@abstractmethod
def load(cls, path: str, force: bool = False):
"""
This method loads a new object from a directory
:param path: this should be a string, and a valid and existing path
:param force: if this is enabled, any error will not crash the saving process,
whether skip the failing object
:return: the loaded object
"""
@property
def save_name(self):
return self.uuid.hex
class Storable(ABC):
@abstractmethod
def save(self, path: str, *args, **kwargs):
"""
This method should be overwritten to provide save implementation.
:param path: this will be created, if it doesn't exist.
"""
@classmethod
@abstractmethod
def load(cls, path: str, *args, **kwargs):
"""
This method should be overwritten to provide load implementation.
:param path: this should be an existing path.
"""
# TODO: remove
class SequentialSave(DirectorySave, ABC):
""" This class saves a sequence of DirectorySave instances, to a directory. """
@staticmethod
def _validate_path(path):
path = abspath(path)
assert isdir(path), ('the given path does not exist', path)
return path
@classmethod
def save_seq(cls, path: str, objects: List[DirectorySave], loader: Type[DirectorySave], force: bool = False) -> List[DirectorySave]:
"""
This method saves all given objects to a save path.
NOTE: This method should be called from self.save().
:param path: this should be a string, and a valid and existing path
:param loader: this should be the class used for loading (here only for assertions)
:param objects: this should be a list of :type self.loader: instances
:param force: if this is enabled, any error will not crash the saving process,
whether skip the failing object
:return: a list with the saved objects
"""
msg = 'could not save object %s (%s)'
path = cls._validate_path(path)
saved_objects = list()
for obj in objects:
assert isinstance(obj, loader), 'invalid object type'
save_path = join(path, obj.save_name)
if not isdir(save_path):
try:
os.mkdir(save_path)
except Exception as e:
log.error(msg, str(obj), 'mkdir failed', exc_info=e)
if force:
continue
raise
try:
obj.save(save_path, force=force)
except Exception as e:
log.error(msg, str(obj), 'save method failed', exc_info=e)
if force:
continue
raise
saved_objects.append(obj)
return saved_objects
@classmethod
def load_seq(cls, path: str, loader: Type[DirectorySave], force: bool = False) -> List[DirectorySave]:
"""
Load a sequence of objects from a directory.
NOTE: This method should only be called from self.load()
:param path: this should be a string, and a valid and existing path
:param loader: this should be the class used for loading
:param force: if this is enabled, any error will not crash the saving process,
whether skip the failing object
:return: all loaded objects
"""
msg = 'could not load object %s (%s)'
path = cls._validate_path(path)
loaded_objects = list()
for subdirectory in os.listdir(path):
try:
data = bytes.fromhex(subdirectory)
except ValueError as e:
log.error('could not load subdirectory, %s is not hexadecimal', subdirectory, exc_info=e)
if force:
continue
raise
try:
obj = loader.load(join(path, subdirectory), force=force)
obj.uuid = uuid.UUID(bytes=data)
loaded_objects.append(obj)
except Exception as e:
log.error(msg, subdirectory, 'load method failed', exc_info=e)
if force:
continue
raise
return loaded_objects
class StorableTypedDict(dict, Storable):
"""
This class can save its content to disk.
It is a subclass from dict, and handles it keys as subdirectories in the save path.
The values are saved by a user defined implementation.
"""
def __init__(self, t: Type[Storable], *args, **kwargs):
"""
:param t: The type of the stored values. This should be user defined.
"""
super(StorableTypedDict, self).__init__(*args, **kwargs)
self._type = t
assert all(map(lambda i: isinstance(i, t), self.values()))
def save(self, path: str, skip: bool = False, *args, **kwargs) -> dict:
"""
Save the entire content of this dictionary to disk.
:param path: a directory to save to.
:param skip: if this is True, failing save calls are not fatal.
:return: a list with all failed save calls.
"""
os.makedirs(path, exist_ok=True)
failed = dict()
for key, value in self.items():
assert isinstance(key, str), 'names should be string'
assert isinstance(value, self._type), f'{value} is not of type {self._type}'
item_path = join(path, key)
os.makedirs(item_path, exist_ok=True)
try:
value.save(item_path)
except Exception as e:
log.error('could not save item %s %s', key, value, exc_info=e)
if skip:
failed[key] = e
continue
raise
return failed
@classmethod
def load(cls, path: str, load_as: Type[Storable] = Storable,
skip: bool = False, *args, **kwargs) -> Union['StorableTypedDict', tuple]:
"""
Load all subdirectories into a dictionary with a given loader.
:param path: a directory to load from.
:param load_as: the loader class. It will be used to load, each individual subdir.
:param skip: if this is True, failing load calls are not fatal.
:return: if skip is True this return a tuple, with
1. the loaded dictionary
2. the failed load calls, stored inside a dictionary
else it only returns the loaded dictionary.
"""
assert isdir(path)
loaded = cls(load_as)
failed = dict()
for directory in os.listdir(path):
dir_path = join(path, directory)
if not isdir(dir_path):
continue
try:
loaded[directory] = load_as.load(dir_path)
except Exception as e:
log.error('could not load item from %s', dir_path)
if skip:
failed[directory] = e
raise
if skip:
return loaded, failed
return loaded
test_storable.py
import os
import unittest
from os.path import *
from functools import partial
from application.storage.storable import Storable, StorableTypedDict
from ...test_utils import *
class DataMixin:
FILE_NAME = 'test.txt'
MESSAGE = 'this is a test'
class CustomStorable(Storable, DataMixin):
def save(self, path: str, *args, **kwargs):
os.makedirs(path, exist_ok=True)
with open(join(path, self.FILE_NAME), 'wt') as f:
f.write(self.MESSAGE)
@classmethod
def load(cls, path: str, *args, **kwargs):
assert isdir(path), 'given path does not exist'
with open(join(path, cls.FILE_NAME), 'rt') as f:
assert f.read() == cls.MESSAGE
return cls()
def __eq__(self, other):
return isinstance(other, self.__class__)
class FailingCustomStorable(CustomStorable):
class CustomException(Exception): ...
def save(self, path: str, *args, **kwargs):
raise self.CustomException
@classmethod
def load(cls, path: str, *args, **kwargs):
raise cls.CustomException
class TestStorableTypedDict(unittest.TestCase):
def test_init(self):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': CustomStorable()}
)
assert 'name1' in storable_dict
assert 'name2' in storable_dict
def test_save(self):
# success
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': CustomStorable()}
)
failed = storable_dict.save(path, skip=False)
assert not failed
files = os.listdir(path)
for key in storable_dict.keys():
assert key in files, '%s is was not created' % key
assert isdir(join(path, key)), '%s is no dir' % key
# fail
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': FailingCustomStorable()}
)
self.assertRaises(
FailingCustomStorable.CustomException,
lambda: storable_dict.save(path, skip=False)
)
# suppressed fail
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': FailingCustomStorable()}
)
try:
failed = storable_dict.save(path, skip=True)
assert len(failed) == 1, failed
assert 'name2' in failed, failed
assert isinstance(failed['name2'], FailingCustomStorable.CustomException)
except FailingCustomStorable.CustomException:
self.fail('the save exception was not handled')
def test_load(self):
# success
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': CustomStorable()}
)
failed = storable_dict.save(path, skip=False)
assert not failed
loaded = StorableTypedDict.load(path, load_as=CustomStorable, skip=False)
assert not isinstance(loaded, tuple)
assert storable_dict == loaded, (storable_dict, loaded)
# failure
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': CustomStorable()}
)
failed = storable_dict.save(path, skip=False)
assert not failed
self.assertRaises(
FailingCustomStorable.CustomException,
lambda: StorableTypedDict.load(path, load_as=FailingCustomStorable, skip=False)
)
# suppressed failure
@tmp_dir
def _(path):
storable_dict = StorableTypedDict(
CustomStorable,
{'name1': CustomStorable(), 'name2': CustomStorable()}
)
failed = storable_dict.save(path, skip=False)
assert not failed
try:
loaded, failed = StorableTypedDict.load(path, load_as=FailingCustomStorable, skip=True)
storable_dict.pop('name2')
assert loaded == storable_dict
assert 'name2' in failed
assert isinstance(failed['name2'], FailingCustomStorable.CustomException)
except FailingCustomStorable.CustomException:
self.fail('the load exception was not handled')
if __name__ == '__main__':
unittest.main()
test_utils.py
import shutil
import tempfile
DELETE_TMP_DIRS = True
def tmp_dir(fn):
path = tempfile.mkdtemp()
try:
return fn(path)
finally:
if DELETE_TMP_DIRS:
shutil.rmtree(path)
