JustAnyone/decman
1
0
Fork 0
mirror of https://github.com/kiviktnm/decman.git synced 2026-09-19 12:08:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Refactor main app

Browse Source
This commit is contained in:
Kivi Kaitaniemi
2025-12-13 17:35:21 +02:00
parent 364bf3d7b9
commit 9471bbf6ac
5 changed files with 215 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/decman/app.py
+187 -132
View File
@@ -28,8 +28,8 @@ def main():
parser.add_argument("--source", action="store", help="python file containing configuration")
parser.add_argument(
"--print",
"--dry-run",
"--print",
action="store_true",
default=False,
help="print what would happen as a result of running decman",
@@ -62,28 +62,27 @@ def main():
with _store.Store(_STORE_FILE, args.dry_run) as store:
try:
_execute_source(store, args)
failed = run_decman(store, args)
except OSError as error:
output.print_error(
f"Unexpected OSError while running decman: {error.strerror or str(error)}"
)
output.print_traceback()
failed = not run_decman(store, args)
except errors.SourceError as error:
output.print_error(f"Error raised manually in the source: {error}")
output.print_traceback()
failed = True
except errors.CommandFailedError as error:
output.print_error(f"{error}")
output.print_traceback()
failed = True
except errors.InvalidOnDisableError as error:
output.print_error(f"Invalid source. {error}")
output.print_traceback()
failed = True
except Exception as error:
output.print_error(f"Unexpected error while running decman: {error}")
output.print_traceback()
failed = True
except OSError as error:
output.print_error(f"Failed to access decman store file '{_STORE_FILE}': {error.strerror}.")
output.print_error("This may cause already completed operations to run again.")
output.print_traceback()
except Exception as error:
output.print_error(f"Unexpected error while running decman: {error}")
output.print_traceback()
finally:
os.chdir(original_wd)
@@ -91,129 +90,18 @@ def main():
sys.exit(1)
def run_decman(store: _store.Store, args: argparse.Namespace) -> bool:
# Ensure execution order is correct. Remove steps not executed now.
execution_order = []
if args.only:
output.print_debug("Argument '--only' is set. Pruning execution steps.")
for step in decman.execution_order:
if step in args.only:
output.print_debug(f"Adding {step} to execution order.")
execution_order.append(step)
else:
execution_order = decman.execution_order
for skip in args.skip:
output.print_debug(f"Skipping step {skip}.")
execution_order.remove(skip)
output.print_debug(f"Execution order is: {', '.join(execution_order)}.")
# Find newly enabled and disabled modules.
store.ensure("enabled_modules", [])
store.ensure("module_on_disable_scripts", {})
new_modules = []
disabled_modules = []
for module in decman.modules:
if module.name not in store["enabled_modules"]:
new_modules.append(module.name)
for module_name in store["enabled_modules"]:
if module_name not in decman.modules:
disabled_modules.append(module_name)
output.print_debug(f"New modules are: {', '.join(new_modules)}.")
output.print_debug(f"Disabled modules are: {', '.join(disabled_modules)}.")
# Disable hooks should be run before anything else because they might depend on packages that
# are going to get removed.
if not args.no_hooks:
output.print_summary("Running 'before update' -hooks.")
for module in decman.modules:
output.print_info(f"Running 'before update' for {module.name}.")
if not args.dry_run:
module.before_update()
if disabled_modules:
output.print_summary("Running 'on disable' -scripts.")
for disabled_module in disabled_modules:
on_disable_script = store["module_on_disable_scripts"].get(disabled_module, None)
if on_disable_script:
output.print_info(f"Running 'on disable' for {disabled_module}.")
if not args.dry_run:
decman.prg([on_disable_script])
store["enabled_modules"].remove(disabled_module)
store["module_on_disable_scripts"].pop(disabled_module)
# Run main execution order
for step in execution_order:
output.print_debug(f"Running step '{step}'.")
match step:
case "files":
if not file_manager.update_files(
store, decman.modules, decman.files, decman.directories, dry_run=args.dry_run
):
return False
case plugin_name:
plugin = decman.plugins.get(plugin_name, None)
if plugin:
plugin.process_modules(store, decman.modules)
if not plugin.apply(store, dry_run=args.dry_run, params=args.params):
return False
else:
output.print_warning(
f"Plugin '{plugin_name}' configured in execution_order\
but not found in available plugins."
)
# On enable and on change should be ran last since they might depend on effects caused by
# execution steps.
if not args.no_hooks:
output.print_summary("Running 'on enable' -hooks.")
for module in decman.modules:
if module.name in new_modules:
output.print_info(f"Running 'on enable' for {module.name}.")
if not args.dry_run:
module.on_enable()
store["enabled_modules"].append(module.name)
try:
script = _module.write_on_disable_script(
module, conf.module_on_disable_scripts_dir
)
if script:
store["module_on_disable_scripts"][module.name] = script
except OSError as error:
output.print_error(
f"Failed to create 'on disable' script for module {module.name}:\
{error.strerror or str(error)}."
)
output.print_warning(
"This script will NOT be created when decman runs the next time."
)
output.print_summary("Running 'on change' -hooks.")
for module in decman.modules:
if module._changed:
output.print_info(f"Running 'on change' for {module.name}.")
if not args.dry_run:
module.on_change()
output.print_summary("Running 'after update' -hooks.")
for module in decman.modules:
output.print_info(f"Running 'after update' for {module.name}.")
if not args.dry_run:
module.after_update()
return True
def _execute_source(store: _store.Store, args: argparse.Namespace):
"""
Runs decman source. May call ``sys.exit(1)`` if user aborts running the source or reading the
source fails.
Raises:
``SourceError``
If code in the source raises this error manually.
``InvalidOnDisableError``
If modules in the source have invalid on_disable functions.
"""
source = store.get("source_file", None)
source_changed = False
@@ -254,3 +142,170 @@ def _execute_source(store: _store.Store, args: argparse.Namespace):
os.chdir(source_dir)
sys.path.append(".")
exec(content)
def run_decman(store: _store.Store, args: argparse.Namespace) -> bool:
"""
Runs decman with the given arguments and a store.
Returns ``True`` if executed succesfully. Otherwise ``False``.
Raises:
``SourceError``
If code in the source raises this error manually.
``CommandFailedError``
If running any command fails.
"""
store.ensure("enabled_modules", [])
store.ensure("module_on_disable_scripts", {})
execution_order = _determine_execution_order(args)
new_modules = _find_new_modules(store)
disabled_modules = _find_disabled_modules(store)
# Disable hooks should be run before anything else because they might depend on packages that
# are going to get removed.
if not args.no_hooks:
_run_before_update(store, args)
_run_on_disable(store, args, disabled_modules)
# Run main execution order
for step in execution_order:
output.print_debug(f"Running step '{step}'.")
match step:
case "files":
if not file_manager.update_files(
store, decman.modules, decman.files, decman.directories, dry_run=args.dry_run
):
return False
case plugin_name:
plugin = decman.plugins.get(plugin_name, None)
if plugin:
plugin.process_modules(store, decman.modules)
if not plugin.apply(store, dry_run=args.dry_run, params=args.params):
return False
else:
output.print_warning(
f"Plugin '{plugin_name}' configured in execution_order\
but not found in available plugins."
)
# On enable and on change should be ran last since they might depend on effects caused by
# execution steps.
if not args.no_hooks:
_run_on_enable(store, args, new_modules)
_run_on_change(store, args)
_run_after_update(store, args)
return True
def _determine_execution_order(args: argparse.Namespace) -> list[str]:
execution_order = []
if args.only:
output.print_debug("Argument '--only' is set. Pruning execution steps.")
for step in decman.execution_order:
if step in args.only:
output.print_debug(f"Adding {step} to execution order.")
execution_order.append(step)
else:
execution_order = decman.execution_order
for skip in args.skip:
output.print_debug(f"Skipping step {skip}.")
execution_order.remove(skip)
output.print_debug(f"Execution order is: {', '.join(execution_order)}.")
return execution_order
def _find_new_modules(store: _store.Store):
new_modules = []
for module in decman.modules:
if module.name not in store["enabled_modules"]:
new_modules.append(module.name)
output.print_debug(f"New modules are: {', '.join(new_modules)}.")
return new_modules
def _find_disabled_modules(store: _store.Store):
disabled_modules = []
for module_name in store["enabled_modules"]:
if module_name not in decman.modules:
disabled_modules.append(module_name)
output.print_debug(f"Disabled modules are: {', '.join(disabled_modules)}.")
return disabled_modules
def _run_before_update(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'before update' -hooks.")
for module in decman.modules:
output.print_info(f"Running 'before update' for {module.name}.")
if not args.dry_run:
module.before_update(store)
def _run_on_disable(store: _store.Store, args: argparse.Namespace, disabled_modules: list[str]):
if not disabled_modules:
return
output.print_summary("Running 'on disable' -scripts.")
for disabled_module in disabled_modules:
on_disable_script = store["module_on_disable_scripts"].get(disabled_module, None)
if on_disable_script:
output.print_info(f"Running 'on disable' for {disabled_module}.")
if not args.dry_run:
decman.prg([on_disable_script])
store["enabled_modules"].remove(disabled_module)
store["module_on_disable_scripts"].pop(disabled_module)
def _run_on_enable(store: _store.Store, args: argparse.Namespace, new_modules: list[str]):
output.print_summary("Running 'on enable' -hooks.")
for module in decman.modules:
if module.name in new_modules:
output.print_info(f"Running 'on enable' for {module.name}.")
if not args.dry_run:
module.on_enable(store)
store["enabled_modules"].append(module.name)
try:
script = _module.write_on_disable_script(
module, conf.module_on_disable_scripts_dir
)
if script:
store["module_on_disable_scripts"][module.name] = script
except OSError as error:
output.print_error(
f"Failed to create 'on disable' script for module {module.name}:\
{error.strerror or str(error)}."
)
output.print_warning(
"This script will NOT be created when decman runs the next time."
)
output.print_warning(
"You should probably investigate the reason for the error. \
Try to fix it, and re-enable this module."
)
def _run_on_change(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'on change' -hooks.")
for module in decman.modules:
if module._changed:
output.print_info(f"Running 'on change' for {module.name}.")
if not args.dry_run:
module.on_change(store)
def _run_after_update(store: _store.Store, args: argparse.Namespace):
output.print_summary("Running 'after update' -hooks.")
for module in decman.modules:
output.print_info(f"Running 'after update' for {module.name}.")
if not args.dry_run:
module.after_update(store)
src/decman/core/file_manager.py
-2
View File
@@ -42,8 +42,6 @@ def update_files(
Returns:
True if all operations completed successfully, False if installation failed.
"""
output.print_summary("Installing files.")
all_checked_files = []
all_changed_files = []
store.ensure("all_files", [])
src/decman/core/module.py
+13 -4
View File
@@ -8,6 +8,7 @@ import typing
import decman.core.error as errors
import decman.core.fs as fs
import decman.core.store as _store
class Module:
@@ -41,35 +42,43 @@ class Module:
_validate_on_disable(f"{cls.__module__}.{cls.__name__}", func)
def before_update(self):
def before_update(self, store: _store.Store):
"""
Override this method to run python code before updating the system.
``store`` can be used to save persistent data between decman runs.
Handle errors within this function. If an error should abort running decman,
raise SourceError or CommandFailedError.
"""
def after_update(self):
def after_update(self, store: _store.Store):
"""
Override this method to run python code after updating the system.
``store`` can be used to save persistent data between decman runs.
Handle errors within this function. If an error should abort running decman,
raise SourceError or CommandFailedError.
"""
def on_enable(self):
def on_enable(self, store: _store.Store):
"""
Override this method to run python code when this module gets enabled.
``store`` can be used to save persistent data between decman runs.
Handle errors within this function. If an error should abort running decman,
raise SourceError or CommandFailedError.
"""
def on_change(self):
def on_change(self, store: _store.Store):
"""
Override this method to run python code after the contents of this module have been
changed in the source.
``store`` can be used to save persistent data between decman runs.
Handle errors within this function. If an error should abort running decman,
raise SourceError or CommandFailedError.
"""
src/decman/plugins/__init__.py
+3 -3
View File
@@ -1,7 +1,7 @@
import importlib.metadata as metadata
import decman.core.module as module
import decman.core.store as cstore
import decman.core.store as _store
class Plugin:
@@ -25,7 +25,7 @@ class Plugin:
return True
def apply(
self, store: cstore.Store, dry_run: bool = False, params: list[str] | None = None
self, store: _store.Store, dry_run: bool = False, params: list[str] | None = None
) -> bool:
"""
Ensures that the state managed by this plugin is present.
@@ -39,7 +39,7 @@ class Plugin:
"""
return True
def process_modules(self, store: cstore.Store, modules: set[module.Module]):
def process_modules(self, store: _store.Store, modules: set[module.Module]):
"""
Processes a module.
"""
tests/test_decman_app.py
+4 -4
View File
@@ -33,16 +33,16 @@ class DummyModule:
self.on_change_called = False
self.after_update_called = False
def before_update(self):
def before_update(self, store):
self.before_update_called = True
def on_enable(self):
def on_enable(self, store):
self.on_enable_called = True
def on_change(self):
def on_change(self, store):
self.on_change_called = True
def after_update(self):
def after_update(self, store):
self.after_update_called = True
@staticmethod