DEV Community

Cover image for LGTM Devlog 26: Python Graphlib DAGs for Quest Stages
Yuan Gao
Yuan Gao

Posted on

LGTM Devlog 26: Python Graphlib DAGs for Quest Stages

Some further cleanup was done, None values have been switched out for sentinel objects, and the dependency injection started last time has been expanded to also deal with the return values, making the main function more trim, and the quests and tests have been adjusted to work better (there was an unfortunate mis-use of class variables previously). The code can be found at commit 1718538

Here's some highlights:

Quest stages

I'm defining a Quest Stage object, whose job is to contain the details that our game needs to interact with the player. This starts with our base Stage ABC. At the moment it just contains this:

class Stage(ABC):
    @property
    @abstractmethod
    def children(cls) -> List[str]:
        """ List of children nodes of this stage """
        return NotImplemented

    quest: Union[Quest, NoQuestType] = NoQuest

    def __repr__(self):
        return f"{self.__class__.__name__}(quest={repr(self.quest)})"
Enter fullscreen mode Exit fullscreen mode

A repr, children property which subclasses need to fill in (indicating the next quest(s) in the graph), and a quest instance property, which will be populated by the quest that spans the stage, which provides this class a parent reference to access things like quest data, or the user object.

From this Stage baseclass, I will implement subclasses for each of the quest stage types. Right now, I only have a debug stage (which doesn't do anything, it's just there for testing), and the definition (but not implementation for) the CreateIssueStage which later will contain implementation for starting a post to GitHub Issues

class DebugStage(Stage):
    """ For debugging purposes """


class CreateIssueStage(Stage):
    """ This stage posts a new issue to a user's fork """
Enter fullscreen mode Exit fullscreen mode

There will be future subclasses that do things like check conditions (to allow branching), check issue replies, and so on. I'll figure out how to do these things later.

Therefore a quest with stages can be defined like this:

class DebugQuest(Quest):
    class QuestDataModel(QuestBaseModel):
        a: int = 1

    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "This is a quest to facilitate testing/debugging"

    class Start(DebugStage):
        children = ["First"]

    class First(DebugStage):
        children = ["Second"]

    class Second(DebugStage):
        children = []
Enter fullscreen mode Exit fullscreen mode

Each quest stage subclasses one of the existing stages, and provide metadata and other values needed by that class (they are not instances)

Collecting stages

During subclassing, the following code runs inside the Quest object:

class Quest(ABC):
    ... 
    @property
    @abstractmethod
    def stages(cls) -> Dict[str, Type[Stage]]:
        """ The initial default data to start quests with """
        return NotImplemented

    def __init_subclass__(cls):
        """ Subclasses instantiate by copying default data """
        # build class list
        cls.stages = {}
        for name, class_var in vars(cls).items():
            if isclass(class_var) and issubclass(class_var, Stage):
                cls.stages[name] = class_var
Enter fullscreen mode Exit fullscreen mode

This initsubclass method is called when Quest is subclassed! And when it does so, the code will collect all the properties of the class, find the ones that are subclasses of Stage and stick them into the class variable stages, so that this dictionary now contains a name: class pair of all the stages. Note: this still doesn't happen during instantiation, so at this point it's still only Classes we're dealing with.

DAGs for quest stages

Each quest stage has a "children" property that defines one or more next stages. This means quests are not just a linear sequence of events, but can be possibly branching ones with things going on simultaneously. This might be overkill for a git learning game, but I can re-use this system for the future.

There's a specify type of graph in graph theory, called a Directed Acyclic Graph, that makes a lot of these quest graphs easier to handle. They're characterised by being directional (which makes sense, since you should only be able to get from one quest stage to another in one direction, quest stages shouldn't send you back one stage); and do not contain cycles. While a quest could by cyclic, I think there's a chance that players can become stuck in the cycle, or the cycle becomes very obvious, with repeating events. So limiting quests to being acyclic has some benefits, and in exchange we have a much easier-to-deal with quest stage layout.

I previously talked a bit about DAGs/DiGraphs in my Advent of Code 2020 Day 10 post, where I used the versatile python NetworkX package to compute the challenge solution very easily.

However, this time, I'm using Python's new graphlib which is new in Python 3.9!

What graphlib and it's TopologicalSorter() allows us to do two very important things:

  1. It allows us to detect disallowed Cycles and loops in the quest. If someone accidentally makes a quest stage loop back on itself, then graphlib will complain (and tests will fail)
  2. It allows us to easily find what the next available quest stage to do is, so we don't have to do any manual graph traversal to find the stages that are ready to run.

The code looks like this, and lives in the Quest ABC:

class Quest(ABC):
    ...

    def load_stages(self) -> None:
        """ loads the stages """

        # load graph
        self.graph = TopologicalSorter()
        for stage_name, StageClass in self.stages.items():
            for child_name in cast(List[str], StageClass.children):
                if child_name not in self.stages:
                    raise QuestDefinitionError(
                        f"Quest {self.__class.__name__} does not have stage named '{child_name}'"
                    )
                self.graph.add(child_name, stage_name)

        try:
            self.graph.prepare()
        except CycleError as err:
            raise QuestDefinitionError(
                f"Quest {self.__class__.__name__} prepare failed! {err}"
            ) from err
Enter fullscreen mode Exit fullscreen mode

The class TopologicalSorter() is provided by Python's built-in graphlib. We instantiate that, and load all the stages from our pre-build stages dictionary, using the stage's name as the node. We're unlikely to have a collision, because they're defined as class properties. This also does some checks for references to non-existent child stages.

Once the TopologicalSorter is loaded, running self.graph.prepare() will perform checks for cycles, and raise exceptions if they exist.

With this function, we can check that all quests are loadable and don't have any cycles in them. We can also do additional checks for things like having one start stages, and for having un-finishable quests, or disconnected stages. These are all classic graph theory problems that we could write later should we want to add checks!

Tests

As per usual, all this new features have a full set of tests, including creation of bad quests, for example a couple of bad quests are defined in the tests files:

class BadStageCycle(Quest):
    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "Bad quest for testing, it has malformed stages"

    class Start(DebugStage):
        children = ["Loop"]

    class Loop(DebugStage):
        """ This should form a cycle, and get flagged by test """

        children = ["Start"]


class BadStageNotExist(Quest):
    version = VersionInfo.parse("1.0.0")
    difficulty = Difficulty.RESERVED
    description = "Bad quest for testing, it has malformed stages"

    class Start(DebugStage):
        """ This references a stage that doesn't exist, and get flagged """

        children = ["Loop"]

def test_fail_stage():
    """ Test bad quests that fail due to stage problems """

    quest = BadStageCycle()
    with pytest.raises(QuestDefinitionError):
        quest.load_stages()

    quest = BadStageNotExist()
    with pytest.raises(QuestDefinitionError):
        quest.load_stages()

Enter fullscreen mode Exit fullscreen mode

And several other tests, which ensure we have good test coverage.


These are some major architectural changes, but things are nice and clean. The next step will be to add a way to actually execute the quest stages in the order that they're loaded.

Top comments (0)