Kumi
e82ccb2701
- Extended copyright to reflect the current year. - Incremented project version to indicate new features and fixes. - Added a new script entry for easier execution, increasing utility and accessibility. - Updated project URLs for better alignment with current infrastructure. - Refactored settings path for simplicity and consistency across deployments. - Improved code readability and maintenance across several modules by cleaning up redundant code, adding missing type annotations, and ensuring consistent code formatting. - Enhanced logging capabilities and error handling to improve diagnostics and troubleshooting, supporting more robust error recovery mechanisms. - Implemented more graceful handling of termination signals to ensure clean shutdown and resource cleanup, enhancing the robustness of the application in production environments. - Introduced command-line argument parsing for configuration file path customization, improving flexibility in different runtime environments. These changes collectively improve the project's maintainability, reliability, and user experience, laying a stronger foundation for future development.
102 lines
3 KiB
Python
102 lines
3 KiB
Python
from .file import File
|
|
|
|
import os
|
|
import pathlib
|
|
|
|
from configparser import SectionProxy
|
|
from typing import Union, Optional
|
|
|
|
|
|
class Directory:
|
|
"""Class representing a Directory on the local filesystem"""
|
|
|
|
@classmethod
|
|
def fromConfig(cls, config: SectionProxy):
|
|
"""Create Directory object from a Directory section in the Config file
|
|
|
|
Args:
|
|
config (configparser.SectionProxy): Configuration section defining
|
|
a Directory
|
|
|
|
Raises:
|
|
ValueError: Raised if section does not contain Location parameter
|
|
|
|
Returns:
|
|
classes.directory.Directory: Directory object for the location
|
|
specified in the config section
|
|
"""
|
|
if "Location" in config.keys():
|
|
return cls(config.name.split()[1], config["Location"])
|
|
else:
|
|
raise ValueError(
|
|
"Definition for Directory "
|
|
+ config.name.split()[1]
|
|
+ " does not contain Location!"
|
|
)
|
|
|
|
def __init__(self, name: str, location: Union[str, pathlib.Path]):
|
|
"""Initialize a new Directory object
|
|
|
|
Args:
|
|
name (str): Name of the Directory object
|
|
location (str, pathlib.Path): Filesystem location of the Directory
|
|
|
|
Raises:
|
|
ValueError: Raised if passed location does not exist or is not a
|
|
directory
|
|
"""
|
|
self.name = name
|
|
|
|
if os.path.isdir(location):
|
|
self.location = pathlib.Path(location)
|
|
self.assertCompletedDirectory()
|
|
else:
|
|
location = str(location)
|
|
raise ValueError(
|
|
f"Location {location} for Directory {name} does not exist or is not a directory."
|
|
)
|
|
|
|
@property
|
|
def completeddir(self):
|
|
return self.location / "completed"
|
|
|
|
def assertCompletedDirectory(self):
|
|
if not os.path.isdir(self.completeddir):
|
|
if os.path.isfile(self.completeddir):
|
|
raise FileExistsError(
|
|
"Cannot create directory %s - path exists but is not a directory!"
|
|
% str(self.completeddir)
|
|
)
|
|
|
|
os.mkdir(self.completeddir)
|
|
|
|
def getFiles(self) -> list[File]:
|
|
"""Get all Files in Directory
|
|
|
|
N.B.: Returns only Files that have a size greater than 0
|
|
|
|
Returns:
|
|
list: List of File objects for files within the Directory
|
|
"""
|
|
files = [
|
|
f
|
|
for f in os.listdir(self.location)
|
|
if os.path.isfile(self.location / f) and os.path.getsize(self.location / f)
|
|
]
|
|
return [File(f, self) for f in files]
|
|
|
|
def getFile(self, name: str) -> Optional[File]:
|
|
"""Get a file in the Directory by name
|
|
|
|
Args:
|
|
name (str): Filename of the File to get
|
|
|
|
Returns:
|
|
File, optional: File object if the file was found, else None
|
|
"""
|
|
|
|
try:
|
|
return File(name, self)
|
|
except FileNotFoundError:
|
|
return None
|