Deconstructing Dependency Injection in my Blog Helper Scripts

For managing my blog I have written a bunch of Python scripts. And over time they have grown. In a frenzy I have refactored it with dependency injection, but now start to find it overengineered and deconstruct that.

When I advocate for design patterns and dependency injection at work, I sometimes get pushback because these ways of designing software feel complicated compared to just writing code in long functions and calling one from the other. I tend to argue that it makes sense to decouple things to increase testability, decrease maintenance costs and so on. So I can already hear people say: “Martin, why aren't you happy with your fancy design patterns in your code? Aren't they supposed to be amazing?”

Well, let me tell you how I was wrong here and how that actually makes me more right somewhere else. The thing is scale. My blog scripts are written and used by exactly one person, myself. At work we are like 30 developers with a code base with order of 100,000 lines of code that is used in a production environment with millions of users. That is a completely different environment.

Another line that I argue along is cohesion vs. coupling. I haven't invented that of course, for instance see Ousterhout¹ for a nice book that features these thoughts. And in a large scale enterprise code base you of course have various components that need to be decoupled. Here in my blog helper scripts, I don't really have the issue of too much coupling. It is just a single component. So I rather have an issue with a lack of cohesion after breaking everything down into the tiniest bits.

Therefore this story can be a tale about when to stop atomizing parts and letting them be. I still advocate for decoupling business logic from data persistance. And I will retain a couple of abstractions here. But I will deconstruct a couple of of them which have escalated.

Superfluous abstractions in the code

There are three banks of things that I have overengineered:

1. Validators. These take a post and perform a single validation on it. This is read-only and will emit a warning. It could be for instance if a post has no preview image but there are image files in the directory. Or when I use a category which doesn't exist and there likely is a typo there.
2. Fixers. These take a post and change it in some way. If there are images which are not links, they are turned to links. If I have an URL with a path to my local preview server, it is changed. I have a couple of tag replacements which are automatically applied.
3. Build actions. These are actions that are performed when I build the blog.

Each one of them has many implementations. For the validators and fixers there usually is an aggregate which can encapsulate multiple things behind one. The following UML diagram shows all of them:

We can have a look at one example, the fixer which performs the tag replacements. I want to use plural terms for tags, and I used “cycle path” as a tag before, I give it the rule “cycle path → cycle paths”. It will take a look into the dictionary of replacements and apply them. The replacements are given during construction time and then it can be used by invoking the fix method with only the post as an argument:

class TagReplacementFixer(Fixer):
    def __init__(self, replacements: dict[str, str]):
        self._replacements = replacements

    def fix(self, post: Post) -> bool:
        was_changed = False
        for source, target in self._replacements.items():
            if source in post.tags:
                post.tags.remove(source)
                post.tags.append(target)
                was_changed = True
        return was_changed

This then allows me to write an aggregate which lets multiple fixers appear like a single one:

class AggregateFixer(Fixer):
    def __init__(self, fixers: Sequence[Fixer]) -> None:
        self._fixers = fixers

    def fix(self, post: Post) -> bool:
        was_changed = False
        for fixer in self._fixers:
            was_changed |= fixer.fix(post)
        return was_changed

And finally cast that into an action that I can call somewhere else:

class ApplyFixesAction(PostsAction):
    def __init__(self, fixer: Fixer) -> None:
        self._fixer = fixer

    def perform(self, posts: list[Post]) -> None:
        logger.info("Applying fixes …")
        for post in posts:
            if self._fixer.fix(post):
                post.write()

This code looks clean because it is completely decoupled. These things do not depend on each other, they only use the Fixer interface. I can put any fixer (or a mock) into the ApplyFixesAction and test it with that. I can either supply an instance of the AggregateFixer or directly one of the concrete fixers. I can test all the fixers independently of the calling code. This might feel nice and clean.

But it falls into the trap that Martin's book Clean Code[^Martin2008] shows: It also feels overengineered for what it really does. One has to be honest here: Do I really need tests for these things? Actually, no. I am the only developer here, and I can see that these fixers work. If they don't, I just correct them. At worst they mess up some blog posts, and since they are all in Git, I revert that. Testability is a burden here, not a feature. It would be a completely different call if my blog scripts were a product in themselves. Or if I worked on it with many other people. But here I don't see the advantage.

[^Martin2008]. Martin, R. C. Clean Code: A Handbook of Agile Software Craftsmanship. (Pearson, 2008).

There is a real cost to this design as well. Let's have a look at the composition root where everything comes together. There we have a big chunk of code which constructs all these things. And it unpacks the stuff from the configuration file.

    actions: list[PostsAction] = []

    if fix:
        actions.append(
            ApplyFixesAction(
                AggregateFixer(
                    [
                        GenderColonFixer(),
                        LocalhostLinkFixer(),
                        ImageLinkFixer(),
                        CategoryFixer(**config["fixers"]["category"]),
                        TagReplacementFixer(**config["fixers"]["tags"]),
                        TeaserFixer(),
                    ]
                )
            )
        )

    if validate:
        actions.append(
            ValidateAction(
                CompositePostValidator(
                    [
                        KnownCategoriesValidator(
                            config["validators"]["known_categories"]
                        ),
                        PreviewImageValidator(),
                        AllFilesUsedValidator(),
                        DeprecatedTagsValidator(
                            config["validators"]["deprecated_tags"]
                        ),
                        SlugValidator(),
                        MathValidator(),
                    ]
                )
            )
        )

And then later we can just call the list of actions on all the posts:

    for action in actions:
        action.perform(all_posts)

So what's the cost here? We have a bunch of code that is a bit cumbersome to read if one is not used to dependency injection. I don't really mind that sort of code, but I don't fit it beautiful either. The real problem here is that this is premature flexibility. Do I ever call these functions in a different order? No. Do I inject some different implementation here sometimes? No. Do I modify the list of actions after I built them, like a TensorFlow graph can be optimized after construction? No. So for what have I built all these abstraction? I don't know. And that's a pretty bad reason to do it.

Deconstructing abstractions

First I want to simplify the validators. I will just write them as simple functions. Let's look at one of the classes. It used to look like this:

class KnownCategoriesValidator(PostValidator):
    def __init__(self, known_categories: list[str]) -> None:
        self._known_categories = known_categories

    def validate(self, post: Post) -> list[str]:
        errors = []
        if post.category not in self._known_categories:
            errors.append(f"Unknown category “{post.category}”")
        return errors

And now it is a simple function:

def validate_known_categories(post: Post, known_categories: list[str]) -> list[str]:
    errors = []
    if post.category not in known_categories:
        errors.append(f"Unknown category “{post.category}”")
    return errors

The interface PostValidator can be removed and the classes CompositePostValidator and ValidateAction restructured. Before we had this code:

class PostValidator(abc.ABC):
    @abc.abstractmethod
    def validate(self, post: Post) -> list[str]:
        raise NotImplementedError()


class CompositePostValidator(PostValidator):
    def __init__(self, validators: list[PostValidator]) -> None:
        self._validators = validators

    def validate(self, post: Post) -> list[str]:
        return [
            warning
            for validator in self._validators
            for warning in validator.validate(post)
        ]


class ValidateAction(PostsAction):
    def __init__(self, validator: PostValidator) -> None:
        self._validator = validator

    def perform(self, posts: list[Post]) -> None:
        logger.info("Validating posts …")
        for post in posts:
            for warning in self._validator.validate(post):
                logger.warning(f"{post.path}: {warning}")

This is the code that replaces all this driver code:

def validate_posts(posts: list[Post], config: dict) -> None:
    for post in posts:
        warnings = []
        warnings += validate_known_categories(post, config["known_categories"])
        warnings += validate_preview_image(post)
        warnings += validate_all_files_used(post)
        warnings += validate_deprecated_tags(post, config["deprecated_tags"])
        warnings += validate_slug(post)
        warnings += validate_math(post)
        for warning in warnings:
            logger.warning(f"{post.path}: {warning}")

We need to pass the part of the config dict into this function. That comes from a section of the TOML file that I use to configure my scripts:

[validation]
known_categories = ["Verkehr", "Reisen", "Code & Zahlen", "Diverses", "Sport"]
deprecated_tags = ["StVO", "Bußgeldstelle", "Verkehrslenkung", "Fahrrad", "Tiefbauamt"]

The block in the main function has also reduced to just a function call and passing the config dict along:

    if validate:
        validate_posts(all_posts, config["validation"])

This is much more compact now. Sure, I have coupled this validate_posts to the configuration format and also to the concrete validation functions. But is that really a problem in my particular application? If I want to extend it, I can just extend it. It's not like I would be providing a plugin architecture for various users.

Log output

In another part of the code I have a bunch of functions that generate things based on the posts. These are all just functions, and I have built a structure of functions and then call them all:

    page_generators = [
        make_update_page,
        write_map_json,
        make_draft_page,
        make_link_graph,
        make_tag_page,
        make_stats,
        make_dev_page,
        generate_whatsapp_page,
    ]

    for page_generator in page_generators:
        logger.info(f"Run generator {page_generator.__name__}() …")
        page_generator(all_posts)

The nice thing here is that we can print the name of each page generator before using it. Yet it obfuscates what is happening in this part of the code because we don't directly see which functions are called after each other.

I have refactored this into something where we just call each function and we're done:

    make_update_page(posts)
    write_map_json(posts)
    make_draft_page(posts)
    make_link_graph(posts)
    make_tag_page(posts)
    make_stats(posts)
    make_dev_page(posts)
    generate_whatsapp_page(posts)

But now we don't have this logging aspect any more. First of all, it isn't really needed. Each one finishes really quickly, and I know what generators I have in there. There is no real point in having this logging. It was just made easy by the structure that I had.

One could always add it back by writing a logging decorator that can be applied to each of these generators:

def log_message(message: str) -> Callable:
    def decorator(function: Callable) -> Callable:
        def wrapped(*args, **kwargs) -> Any:
            logger.info(message)
            return function(*args, **kwargs)
        return wrapped
    return decorator

And then I can just use that with @log_message("Generate update page") when I want to highlight the execution of that function. This then actually uses the decorator pattern with the @-syntax in Python to put the logging closer to the function while not coupling it to it.

Extracting a publishing strategy

One thing that still annoys me due to a lack of abstraction is the publishing of posts. I currently have this function which determines the posts to publish as well as publishing them:

def auto_publish(posts: list[Post], threshold: int) -> None:
    logger.info("Publishing posts …")
    today = datetime.date.today()
    num_posts_published_today = sum(
        post.date == today for post in posts if post.status == PostState.published
    )
    ready_posts = [
        post
        for post in posts
        if post.status == PostState.draft
        and post.board == BoardState.ready
        and post.date < today
    ]
    ready_posts.sort(key=lambda post: (post.date, post.path))
    num_ready = len(ready_posts)
    num_to_publish = (num_ready + threshold - 1) // threshold
    print(f"There are {len(ready_posts)} posts in the ready queue.")
    print(f"Published today: {num_posts_published_today}")
    print(f"Target for today: {num_to_publish}")
    while num_posts_published_today < num_to_publish and ready_posts:
        post = ready_posts.pop(0)
        print(f"Publishing: {post.title}")
        publish_post(post)
        num_posts_published_today += 1

I have written about the number of articles per day and the emerging structure of my queue. The problem is that this was just emergent and that I don't have this concept in code. Recently I wanted to have a calendar of posts in the queue and wrote a redundant implementation:

def get_planned_posts(
    posts: list[Post], threshold: int
) -> dict[datetime.date, list[Post]]:
    posts.sort()
    today = datetime.date.today()
    future_posts = [post for post in posts if post.date >= today and not post.is_draft]
    ready_posts = [
        post for post in posts if post.is_draft and post.board == BoardState.ready
    ]
    day = today
    planner: dict[datetime.date, list[Post]] = {}
    while future_posts or ready_posts:
        future_posts = [
            post for post in posts if post.date >= day and not post.is_draft
        ]
        posts_this_day = [
            post for post in future_posts if post.date == day and not post.is_draft
        ]
        ready_since_day_before = [post for post in ready_posts if post.date < day]
        num_to_publish = (len(ready_since_day_before) + threshold - 1) // threshold
        while len(posts_this_day) < num_to_publish and ready_posts:
            posts_this_day.append(ready_posts.pop(0))
        planner[day] = posts_this_day
        day += datetime.timedelta(days=1)
    return planner

The second function can function as a strategy for the first one. And using that makes the auto_publish function much more compact:

def auto_publish(posts: list[Post], threshold: int) -> None:
    logger.info("Publishing posts …")
    planner = get_planned_posts(posts, threshold)
    today = datetime.date.today()
    for post in planner[today]:
        if post.is_draft:
            print(f"Publishing: {post.title}")
            publish_post(post)

Now we only have one instance of the publishing strategy. If I desire to have a different strategy, I could just change that single function or add a second function and replace it. There is no strict need to define an interface here. Having a function that returns a dict that maps dates to a list of posts is completely sufficient for my needs.

Conclusion

Having too much abstraction in the code is a hindrance to working with it. It is different from having too little abstraction. It is important to find a sensible spot where there is not too much and not too little abstraction. If one finds out that one has gone too far into one direction it is sensible to just return to the sweet spot.