from importlib import import_module
from graphlib import TopologicalSorter
from typing import Iterable
from django.apps import apps, AppConfig
from ox.core.apps import AppConfig as OxAppConfig
__all__ = (
"get_or_create_app_config",
"order_apps_dependencies",
"DiscoverModules",
)
[docs]
def get_or_create_app_config(module: str) -> AppConfig:
"""Get AppConfig based on app module.
It looks up in apps registry, if not found tries to import it (using ``AppConfig.create``).
:param module: app module path
:return: the AppConfig instance
:raises ImportError: module couldn't be imported;
:raises ImproperlyConfigured: app was not configured properly;
"""
try:
return apps.get_app_config(module)
except (KeyError, LookupError):
return AppConfig.create(module)
[docs]
def order_apps_dependencies(app_configs: None | Iterable[AppConfig | str] = None) -> list[AppConfig]:
"""
Return applications ordered by dependencies (topological sort).
:param app_configs: an iterable of AppConfig to look dependencies for (defaults to apps registry)
:return: a list of ordered app configs by dependencies.
"""
app_configs = list(app_configs or reversed(apps.get_app_configs()))
graph = TopologicalSorter()
done = {}
for app in app_configs:
if isinstance(app, str):
if app in done:
continue
app = get_or_create_app_config(app)
if app.name in done:
continue
if deps := getattr(app, "dependencies", None):
app_configs.extend(deps)
graph.add(app.name, *deps)
else:
graph.add(app.name)
done[app.name] = app
return [done[name] for name in graph.static_order()]
[docs]
class DiscoverModules:
"""Utility function used to discover sub-modules in applications.
Basically you provide lookup module names and handler methods.
When calling :py:meth:`run` or :py:meth:`run_handler`, the applications
will be scanned to find the corresponding module(s). For each, a
corresponding handler will be called.
For each declared sub-module, there must be an equivalent method
handler with the following signature: ``handle_{module_name}(self,
app, module, **kw)`` (where dots in ``module_name`` are replace by ``_``)
Example:
.. code-block::
discover = DiscoverModules(
["urls", "nested.module"],
handle_urls = lambda app, mod, **kw: print(app, mod, kw),
handle_nested_module = lambda app, mod, **kw: print(app, mod, kw)
)
# Running handlers
discover.run(foo="bar") # for all searched modules
discover.run_handler("urls", foo="bar") # for a single module
# Or by subclassing
class Discover(DiscoverModule):
module_names = "urls"
oxylus_apps = True
# Named parameters are provided by user when calling run or
# run_handler.
def handle_urls(self, app, module, **kwargs):
print(app, module, **kwargs)
"""
module_names: str | Iterable[str] = ""
"""A single or a list of module names to look-up for."""
oxylus_apps: bool = False
"""Only run over Oxylus applications (subclasses of :py:class:`ox.core.apps.AppConfig`). """
def __init__(self, module_names: str | Iterable[str] | None = None, oxylus_apps: bool = False, **handlers):
"""
:param module_names: list of module names to look up
:param oxylus_apps: only run over Oxylus applications
:param handlers: list of handler functions.
"""
if module_names is not None:
self.module_names = module_names
self.oxylus_apps = oxylus_apps
if handlers:
for key, handler in handlers.items():
if not key.startswith("handle_"):
raise ValueError(f"Invalid handler method name: {key} (should start with `handle_`)")
setattr(self, key, handler)
[docs]
def run(self, app_configs: Iterable[AppConfig] | None = None, **kw):
"""Run handler over all modules."""
app_configs = self.get_app_configs(app_configs)
if app_configs is None:
app_configs = apps.get_app_configs()
for module_name in self.get_module_names():
self.run_handler(module_name, app_configs, **kw)
[docs]
def run_handler(self, module_name: str, app_configs: Iterable[AppConfig] = None, **kw):
"""Run handler over app config looking for the provided module."""
handler = self.get_handler(module_name)
app_configs = self.get_app_configs(app_configs)
for app in app_configs:
if mod := self.get_app_module(app, module_name):
handler(app, mod, **kw)
[docs]
def get_app_configs(self, app_configs: Iterable[AppConfig] | None = None):
"""Return applications configs, filtered based on attributes."""
if app_configs is None:
app_configs = apps.get_app_configs()
if self.oxylus_apps:
app_configs = [a for a in app_configs if isinstance(a, OxAppConfig)]
return app_configs
[docs]
def get_module_names(self) -> Iterable[str]:
"""Return modules names as an iterable"""
if isinstance(self.module_names, str):
return [self.module_names]
return self.module_names
[docs]
def get_handler(self, module_name):
"""Return handler function for the provided"""
fn = f"handle_{module_name.replace('.','_')}"
if not hasattr(self, fn):
raise NotImplementedError(f"Handler is not implemented for `{module_name}`. Expected `{fn}`.")
return getattr(self, fn)
[docs]
def get_app_module(self, app, module_name):
"""Yield `app, sub_module` for each sub-module found for apps."""
path = f"{app.name}.{module_name}"
try:
return import_module(path)
except ModuleNotFoundError as err:
if err.msg == f"No module named '{path}'":
return None
raise