Skip to content

Core#

cli #

Main module to expose internal functions as command lines.

renameit(config_path=None, config_dir=None) #

This is the main API that is used to trigger renaming jobs.

Parameters:

Name Type Description Default
config_path Optional[str]

File path to configs overriding the default path. Defaults to None.

 None
config_dir Optional[str]

Dir path to other config files if needed. Defaults to None.

 None
Source code in renameit/core/cli.py 
11
12
13
14
15
16
17
18
19
20
21
22
def renameit(config_path: Optional[str] = None, config_dir: Optional[str] = None):
    """This is the main API that is used to trigger renaming jobs.

    Args:
        config_path: File path to configs overriding the default path. Defaults to None.
        config_dir: Dir path to other config files if needed. Defaults to None.
    """
    config = ProjectConfig.create_new(config_path=config_path, config_dir=config_dir)

    job_runner = JobRunnerStandard(jobs=config.get_jobs())

    job_runner.execute()

job_runners #

This module contains job runner classes that will run the renaming jobs

The module contains the following:

  • Job - Class representing renaming job.
  • JobRunnerStandard - Implementation for standard job runner.

Job #

The job class that contains the processing details. Providing the central point for file systems and handlers interactions.

Parameters:

Name Type Description Default
name str

Name of the job.

 required
file_system IFileSystem

Which file system to use.

 required
rename_handler IHandler

Which renaming handler to use.

 required
operation str

What operation to do on the matched files, accepted values are (rename, copy) case sensitive.

 required
Source code in renameit/core/job_runners.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
class Job:
    """The job class that contains the processing details. Providing the central point
    for file systems and handlers interactions.

    Args:
        name (str): Name of the job.
        file_system (IFileSystem): Which file system to use.
        rename_handler (IHandler): Which renaming handler to use.
        operation (str): What operation to do on the matched files,
                        accepted values are (rename, copy) case sensitive.
    """

    def __init__(
        self,
        name: str,
        file_system: IFileSystem,
        rename_handler: IHandler,
        operation: str,
    ):

        if (
            file_system.__class__.copy_object == IFileSystem.copy_object
            and operation == "copy_object"
        ):
            raise NotImplementedError(
                f"FileSystem type `{file_system.name}` does not support copying objects."
            )
        if (
            file_system.__class__.rename_object == IFileSystem.rename_object
            and operation == "rename_object"
        ):
            raise NotImplementedError(
                f"FileSystem type `{file_system.name}` does not support renaming objects."
            )

        self.name = name
        self.file_system = file_system
        self.rename_handler = rename_handler
        self.operation = operation

    def run(self):
        """A job needs to run at some point."""

        logging.info(f"Running job {self.name}")

        for file_obj in self.file_system.list_objects():
            new_file_obj = self.rename_handler.get_new_name(file_obj=file_obj)

            if file_obj.path != new_file_obj.path:
                getattr(self.file_system, self.operation)(
                    source_obj=file_obj,
                    target_obj=self.rename_handler.get_new_name(file_obj=file_obj),
                )

run() #

A job needs to run at some point.

Source code in renameit/core/job_runners.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
def run(self):
    """A job needs to run at some point."""

    logging.info(f"Running job {self.name}")

    for file_obj in self.file_system.list_objects():
        new_file_obj = self.rename_handler.get_new_name(file_obj=file_obj)

        if file_obj.path != new_file_obj.path:
            getattr(self.file_system, self.operation)(
                source_obj=file_obj,
                target_obj=self.rename_handler.get_new_name(file_obj=file_obj),
            )

JobRunnerStandard #

A simple implementation for executing jobs in sequence.

Source code in renameit/core/job_runners.py 
90
91
92
93
94
95
class JobRunnerStandard(IJobRunner):
    """A simple implementation for executing jobs in sequence."""

    def execute(self) -> None:
        for job in self.jobs:
            job.run()

Config#

project #

Config classes, simply extending the pydantic BaseModel are defined here to provide a contract for users to abide by. Schema definition, validation, serialization and deserialization are all handled within these classes.

We can expose real object instances from within these contracts/configs that can be used later by the app logic in other modules and functions.

FileSystemConfig #

FileSystemConfig.

Source code in renameit/core/config/project.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class FileSystemConfig(BaseModel):
    """FileSystemConfig."""

    type: str
    args: Dict[str, Any]

    @validator("type")
    def type_validation(cls, value) -> str:
        if value not in file_systems_map.keys():
            raise ValueError(
                f"Unknow type `{value}`, only allowed: {', '.join(file_systems_map.keys())}"
            )
        return value

    def get_file_system(self) -> IFileSystem:
        return file_systems_map[self.type](**self.args)  # type: ignore

JobConfig #

JobConfig.

Source code in renameit/core/config/project.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class JobConfig(BaseModel):
    """JobConfig."""

    name: str
    operation: Literal["rename", "copy"]
    file_system_config: FileSystemConfig = Field(alias="file_system")
    rename_handler_config: RenameHandlerConfig = Field(alias="rename_handler")

    def get_job(self) -> Union[Job, None]:
        try:
            return Job(
                name=self.name,
                file_system=self.file_system_config.get_file_system(),
                rename_handler=self.rename_handler_config.get_handler(),
                operation=f"{self.operation}_object",
            )
        except (NotImplementedError, TypeError) as e:
            logging.warning(f"Skipping job {self.name}: {str(e)}")

            if isinstance(e, TypeError):
                logging.exception(str(e))

            return None

ProjectConfig #

UserProjectConfig.

Source code in renameit/core/config/project.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class ProjectConfig(BaseModel):
    """UserProjectConfig."""

    version: str = "0.1"
    jobs: List[JobConfig] = Field(default_factory=list)

    @classmethod
    def create_new(cls, config_path, config_dir=None) -> "ProjectConfig":
        return cls(**load_config(user_file_path=config_path, user_dir_path=config_dir))

    def get_jobs(self) -> List[Job]:
        """Get a list of Job objects from their corresponding list of JobConfig.

        Returns:
            List[Job]: List of Job objects
        """
        return [job for job_config in self.jobs if (job := job_config.get_job()) is not None]
get_jobs() #

Get a list of Job objects from their corresponding list of JobConfig.

Returns:

Type Description
List[Job]

List[Job]: List of Job objects
Source code in renameit/core/config/project.py 
33
34
35
36
37
38
39
def get_jobs(self) -> List[Job]:
    """Get a list of Job objects from their corresponding list of JobConfig.

    Returns:
        List[Job]: List of Job objects
    """
    return [job for job_config in self.jobs if (job := job_config.get_job()) is not None]

RenameHandlerConfig #

RenameHandlerConfig.

Source code in renameit/core/config/project.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class RenameHandlerConfig(BaseModel):
    """RenameHandlerConfig."""

    type: str
    args: Dict[str, Any]

    @validator("type")
    def type_validation(cls, value) -> str:
        if value not in handlers_map.keys():
            raise ValueError(
                f"Unknow type `{value}`, only allowed: {', '.join(handlers_map.keys())}"
            )
        return value

    def get_handler(self) -> IHandler:
        return handlers_map[self.type](**self.args)  # type: ignore

parse #

This module acts as the center point for reconciling user input from config files, env vars and other external configuration services.

Everything related to loading, parsing and rendering configurations can be implemented here.

The result data can be used as input for config classes like the ones in project

load_config(user_file_path=None, user_dir_path=None) #

Load configurations from files.

Parameters:

Name Type Description Default
user_file_path str

Path to the base config file. Defaults to None.

 None
user_dir_path str

Path to the extension config directory. Defaults to None.

 None

Returns:

Type Description
OrderedDict

Single dict with all the loaded config
Source code in renameit/core/config/parse.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def load_config(user_file_path: str = None, user_dir_path: str = None) -> OrderedDict:
    """Load configurations from files.

    Args:
        user_file_path (str, optional): Path to the base config file. Defaults to None.
        user_dir_path (str, optional): Path to the extension config directory. Defaults to None.

    Returns:
        Single dict with all the loaded config
    """
    kwargs = {}
    if user_file_path is not None:
        kwargs.update({"user_file_path": user_file_path})

    if user_dir_path is not None:
        kwargs.update({"user_dir_path": user_dir_path})

    config = confight.load_user_app("renameit", **kwargs)

    return config