I'm working on a replacement for make written in Python. Instead of a Makefile, you create a Wormfile.py. This file gets imported dynamically and a Context object is passed to which you can add your goals (akin to targets in make). Contexts can hold variables and even inherit variables from parent contexts.
Goal is an abstract base class. At a minimum, a subclass needs to implement the exists method. It can also implement compute_last_built which returns the time when the goal was last built (None if the goal doesn't exist or if a build time doesn't make sense for this goal).
What follows is just the definition of Goal and Context. The code assumes Python 3.10+.
What I'm looking for primarily is the identification of any edge cases. Is there something I didn't consider that's going to bite me later?
It may seem superfluous to implement __hash__ and __eq__ in ways that mimic what object already does. However, other parts of my project require these particular implementations and I want typing.final to warn the user not to override them.
from __future__ import annotations
import abc
import collections.abc
import importlib.util
import logging
import os
import pathlib
import typing
_logger = logging.getLogger("sandworm.core")
T = typing.TypeVar("T", bound="Goal")
Builder = collections.abc.Callable[[T], bool]
class Goal(abc.ABC):
"""Goal to be built.
Attributes:
name (str): Name of the goal. Used in logging.
context (Context): Associated context.
"""
def __init__(self: T, name: str, builder: Builder | None = None) -> None:
"""
Arguments:
name (str): Name of the goal.
builder (Callable[[Goal], bool] | None): Function to build the goal.
"""
self._name = name
self._builder = builder
self._dirty = False
self._ctx: Context | None = None
self._dependencies: list[Goal] = []
self._build_time: float | int | None = None
self._build_time_calculated = False
def __repr__(self) -> str:
return self._name
@typing.final
def __eq__(self, other: object) -> bool:
return self is other
@typing.final
def __hash__(self) -> int:
return id(self)
@property
def name(self) -> str:
return self._name
@typing.final
@property
def context(self) -> Context:
if self._ctx is None:
raise Exception("Context has not been set.")
return self._ctx
@typing.final
def dependencies(self) -> collections.abc.Iterator[Goal]:
"""Iterate over the dependencies.
Returns:
Iterator[Goal]
"""
yield from self._dependencies
@typing.final
def add_dependency(self, dependency: Goal) -> None:
"""Add a dependency.
Arguments:
goal (Goal): Dependency.
"""
if self._ctx is not None:
dependency.set_context(self._ctx)
self._dependencies.append(dependency)
@typing.final
def set_context(self, ctx: Context) -> None:
"""Set the context (if unset) on this goal and all of its dependencies (recursively).
Arguments:
ctx (Caontext)
"""
if self._ctx is None:
self._ctx = ctx
for dependency in self._dependencies:
dependency.set_context(ctx)
@abc.abstractmethod
def exists(self) -> bool:
"""Does the goal already exist?
Returns:
bool
"""
raise NotImplementedError
def compute_last_built(self) -> float | int | None:
"""Determine the last time the goal was built.
Returns:
float | int | None: Unix epoch timestamp or `None` if the goal doesn't exist or if a build time
doesn't make sense.
"""
return None
@typing.final
def needs_building(self) -> bool:
"""Does the goal need to be built?
Returns:
bool
"""
last_built = self._last_built()
for dependency in self._dependencies:
if dependency._dirty:
_logger.debug(f"{self} needs to be built because dependency {dependency} is dirty")
return True
if (
last_built is not None
and (dep_last_built := dependency._last_built()) is not None
and dep_last_built > last_built
):
_logger.debug(f"{self} needs to be built because dependency {dependency} is newer")
return True
if missing := not self.exists():
_logger.debug(f"{self} needs to be built because it does not exist")
return missing
@typing.final
def build(self) -> bool:
"""Build the goal.
This method should not be called by the user directly. Instead, run `sandworm.build(goal)`.
Returns:
bool: Was the build successful?
Raises:
Exception: The context has not been set.
"""
if self._ctx is None:
raise Exception("Context is not set.")
if not self._builder:
if not self.exists():
_logger.error(f"No logic exists to build {self}")
return False
self._finish()
return True
_logger.info(f"Building {self}")
if success := self._builder(self):
_logger.debug(f"{self} built successfully")
else:
_logger.error(f"Failed to build {self}")
self._finish()
return success
def _last_built(self) -> float | int | None:
if self._build_time_calculated:
return self._build_time
return self._recompute_last_built()
def _recompute_last_built(self) -> float | int | None:
self._build_time = self.compute_last_built()
self._build_time_calculated = True
return self._build_time
def _finish(self) -> None:
self._dirty = True
self._recompute_last_built()
class FileGoal(Goal):
"""Represents a file to be created.
Attributes:
path (pathlib.Path): Path to the file.
"""
def __init__(self: T, path: str | pathlib.Path, builder: Builder | None = None) -> None:
if isinstance(path, str):
name = path
path = pathlib.Path(path)
else:
name = str(path)
super().__init__(name, builder)
self._path = path
@property
def path(self) -> pathlib.Path:
return self._path
def exists(self) -> bool:
return self._path.exists()
def compute_last_built(self) -> float | None:
return self._path.stat().st_mtime if self._path.exists() else None
class ThinGoal(Goal):
"""Goal that always registers as existing.
This is meant merely to be a wrapper around its dependencies as it is only built if one or more of its
dependencies are dirty.
"""
def exists(self) -> bool:
return True
@typing.final
class Context:
"""Build context.
Attributes:
basedir (pathlib.Path): Base directory for its goals.
main_goal (Goal | None): If not `None`, then the goal that will be run when no goal is
specified for `sandworm.build`.
"""
def __init__(self, directory: str | pathlib.Path, *, parent: Context | None = None) -> None:
if isinstance(directory, str):
directory = pathlib.Path(directory)
if not directory.is_dir():
raise NotADirectoryError(directory)
self._basedir = directory.resolve()
self._parent = parent
self._children: list[Context] = []
self._main_goal: Goal | None = None
self._variables: dict[str, str] = {}
self._goals: dict[str, Goal] = {}
self._cleaners: list[collections.abc.Callable[[Context], None]] = []
if parent is not None:
parent._children.insert(0, self)
@classmethod
def from_directory(cls, directory: str | pathlib.Path, parent: Context | None = None) -> Context:
"""Create a context by loading a Wormfile.
Arguments:
directory (str | pathlib.Path): Directory containing Wormfile.py.
parent (sandworm.Context | None): Optional parent from which the new context will inherit.
Returns:
sandworm.Context
Raises:
FileNotFoundError: The Wormfile could not be found.
ImportError: The Wormfile couldn't be loaded.
"""
if isinstance(directory, str):
directory = pathlib.Path(directory)
wormfile = directory / "Wormfile.py"
if not wormfile.is_file():
raise FileNotFoundError(wormfile)
_logger.debug(f"Loading {wormfile}")
spec = importlib.util.spec_from_file_location("Wormfile", wormfile)
if spec is None:
raise ImportError(str(wormfile))
module = importlib.util.module_from_spec(spec)
if spec.loader is None:
raise ImportError(str(wormfile))
spec.loader.exec_module(module)
ctx = Context(directory, parent=parent)
module.setup_context(ctx)
return ctx
@property
def basedir(self) -> pathlib.Path:
return self._basedir
@property
def main_goal(self) -> Goal | None:
return self._main_goal
def create_child(self) -> Context:
"""Create a child context with the same base directory.
Returns:
Context
"""
return Context(self._basedir, parent=self)
def variables(self) -> dict[str, str]:
"""Iterate over the exposed variables.
This includes any ancestor context's variables but not environment variables.
Returns:
Iterator[tuple[str, str]]: Iterator of key/value pairs.
"""
variables = self._parent.variables() if self._parent is not None else {}
variables.update(self._variables)
return variables
def add_goal(self, goal: Goal, *, name: str | None = None, main: bool = False) -> None:
"""Add a goal to be managed by this context.
Arguments:
goal (Goal): Goal to be added.
name (str | None): Name to reference the goal by. If `None`, then the goal's own name will be
used.
main (bool): Is this to be the context's main goal?
Raises:
ValueError: A duplicate goal name was specified.
"""
if name is None:
name = goal.name
if name in self._goals:
raise ValueError(f"Duplicate goal name: {name}")
goal.set_context(self)
self._goals[name] = goal
if main:
self._main_goal = goal
def lookup_goal(self, name: str) -> Goal | None:
"""Look up a registered goal by name.
Arguments:
name (str)
Returns:
Goal | None
"""
return self._goals.get(name)
def goals(self) -> collections.abc.Iterator[tuple[str, Goal]]:
"""Iterate over the registered goals.
Returns:
Iterator[tuple[str, Goal]]: Iterator of names and goals.
"""
yield from self._goals.items()
def add_cleaner(self, cleaner: collections.abc.Callable[[Context], None]) -> None:
"""Register a cleaner function to be called by `self.clean()`.
Arguments:
Callable[[Context], None]
"""
self._cleaners.insert(0, cleaner)
def clean(self) -> None:
"""Runs all of the registered cleaners.
All child contexts will be cleaned first (in the reverse order that they were added). For each
context, the cleaners will be called in the reverse order that they were added.
"""
for child in self._children:
child.clean()
for cleaner in self._cleaners:
cleaner(self)
def get(self, key: str, default: str | None = None) -> str | None:
"""Look up a variable safely.
First, this context is searched. If the variable is not found, then the parent is searched (if there
is one) and so on. At the end, if the variable is still not found, the environment is searched.
Arguments:
key (str): Name of the variable.
default (str | None): Value to return if the key isn't found.
Returns:
str | None: The value of the variable if it was found and the default otherwise.
"""
if (value := self._variables.get(key)) is not None:
return value
if self._parent is not None:
return self._parent.get(key, default)
return os.environ.get(key, default)
def __contains__(self, key: str) -> bool:
"""Is the variable set?
Like `get`, the ancestry and environment are included in the search.
Arguments:
key (str): Name of the variable.
Returns:
bool
"""
return self.get(key) is not None
def __getitem__(self, key: str) -> str:
"""Look up a variable.
Like `get`, the ancestry and environment are included in the search.
Arguments:
key (str): Name of the variable.
Returns:
str: Vale of the variable.
Raises:
KeyError: The variable was not found.
"""
if (value := self.get(key)) is None:
raise KeyError(key)
return value
def __setitem__(self, key: str, value: str) -> None:
"""Set a variable.
Arguments:
key (str): Name of the variable.
value (str): Value of the variable.
"""
self._variables[key] = value
def set_if_unset(self, key: str, value: str) -> None:
"""Set a variable only if it hasn't already been set.
Like `get`, the ancestry and environment are included in the search.
Arguments:
key (str): Name of the variable.
value (str): Value of the variable.
"""
if key not in self:
self[key] = value
