Skip to content

Commit

Permalink
refacto: remove unused methods and improve doctsrings
Browse files Browse the repository at this point in the history
  • Loading branch information
Guts committed Feb 22, 2024
1 parent 68bcbc2 commit 46b884a
Showing 1 changed file with 9 additions and 32 deletions.
41 changes: 9 additions & 32 deletions qgis_deployment_toolbelt/jobs/orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,8 @@
# ##################################

# Standard library
import importlib
import logging
from os import environ
from pathlib import Path

# project
from qgis_deployment_toolbelt.jobs.generic_job import GenericJob
Expand Down Expand Up @@ -47,7 +45,7 @@
class JobsOrchestrator:
"""Orchestrate jobs."""

JOBS = (
JOBS: tuple = (
JobEnvironmentVariables,
JobPluginsDownloader,
JobPluginsSynchronizer,
Expand All @@ -58,6 +56,7 @@ class JobsOrchestrator:
PACKAGE_NAME: str = "qgis_deployment_toolbelt.jobs"

def __init__(self) -> None:
"""Instanciate orchestrator."""
# log environment variables prefixed with QDT_
qdt_env_vars = {
env_var: value
Expand All @@ -70,20 +69,15 @@ def __init__(self) -> None:
logger.debug(f"{var_name}={var_value}")

@property
def available_jobs(self) -> list[Path]:
"""List all available jobs."""
# job modules
return list(Path(__file__).parent.glob("job_*.py"))

@property
def jobs_ids(self) -> tuple[str]:
def jobs_ids(self) -> tuple[str, ...]:
"""Returns available jobs ID.
:return Tuple[str]: jobs ids
Returns:
tuple[str]: tuple of jobs ids
"""
return tuple([job.ID for job in self.JOBS])

def get_job_module_from_id(self, job_id: str) -> object:
def get_job_module_from_id(self, job_id: str) -> GenericJob | None:
"""Get job class from id.
:param str job_id: job identifier (as defined in scenario file)
Expand All @@ -93,30 +87,13 @@ def get_job_module_from_id(self, job_id: str) -> object:
if job.ID == job_id:
return job

def init_job_class_from_id(self, job_id: str, options: dict) -> GenericJob:
def init_job_class_from_id(self, job_id: str, options: dict) -> GenericJob | None:
"""Get job class from id and instanciate it with options.
:param str job_id: job identifier (i.e. "qprofiles-manager")
:param dict options: job options
:return object: instanciated job class with options
"""
return self.get_job_module_from_id(job_id)(options)

def import_job(self, job_name: str):
"""Import a job."""
return importlib.import_module(name=job_name, package=self.PACKAGE_NAME)

def import_all_jobs(self) -> None:
"""Import all jobs."""
for module in self.available_jobs:
self.JOBS[module.stem] = self.import_job(module.stem)


# #############################################################################
# ##### Stand alone program ########
# ##################################

if __name__ == "__main__":
"""Standalone execution."""
pass
if job := self.get_job_module_from_id(job_id):
return job(options)

0 comments on commit 46b884a

Please sign in to comment.