Add registry pattern

pythonic-patterns/registry_pattern/before/main.py

@@ -0,0 +1,32 @@
+import json
+from typing import Protocol
+
+from tasks import Pulse, Recalibrate, Reinforce
+
+
+class Task(Protocol):
+    def run(self) -> None:
+        ...
+
+
+def main() -> None:
+    with open("./tasks.json", encoding="utf-8") as file:
+        data = json.load(file)
+
+        tasks: list[Task] = []
+
+        for item in data["tasks"]:
+            if item["type"] == "pulse":
+                tasks.append(Pulse(item["strength"]))
+            elif item["type"] == "recalibrate":
+                tasks.append(Recalibrate(item["target"]))
+            elif item["type"] == "reinforce":
+                tasks.append(Reinforce(item["plating_type"], item["target"]))
+
+        # run the tasks
+        for task in tasks:
+            task.run()
+
+
+if __name__ == "__main__":
+    main()

pythonic-patterns/registry_pattern/before/tasks.json

@@ -0,0 +1,17 @@
+{
+  "tasks": [
+    {
+      "type": "pulse",
+      "strength": 190
+    },
+    {
+      "type": "recalibrate",
+      "target": "Thoron subspace transponder"
+    },
+    {
+      "type": "reinforce",
+      "plating_type": "biogenic",
+      "target": "the deflector array"
+    }
+  ]
+}

pythonic-patterns/registry_pattern/before/tasks.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Pulse:
+    strength: int
+
+    def run(self) -> None:
+        print(
+            f"Sending a subspace pulse of {self.strength} microPicards to the converter assembly."
+        )
+
+
+@dataclass
+class Recalibrate:
+    target: str
+
+    def run(self) -> None:
+        print(f"Recalibrating the {self.target}.")
+
+
+@dataclass
+class Reinforce:
+    plating_type: str
+    target: str
+
+    def run(self) -> None:
+        print(f"Reinforcing {self.plating_type} plating of {self.target}.")

pythonic-patterns/registry_pattern/class_based/main.py

@@ -0,0 +1,28 @@
+import json
+
+from registry import TaskRegistry
+from tasks import PulseFactory, RecalibrateFactory, ReinforceFactory
+
+
+def main() -> None:
+
+    # register a couple of tasks
+    task_registry = TaskRegistry()
+    task_registry.register("pulse", PulseFactory())
+    task_registry.register("recalibrate", RecalibrateFactory())
+    task_registry.register("reinforce", ReinforceFactory())
+
+    # read data from a JSON file
+    with open("./tasks.json", encoding="utf-8") as file:
+        data = json.load(file)
+
+        # create the tasks
+        tasks = [task_registry.create(item) for item in data["tasks"]]
+
+        # run the tasks
+        for task in tasks:
+            task.run()
+
+
+if __name__ == "__main__":
+    main()

pythonic-patterns/registry_pattern/class_based/registry.py

@@ -0,0 +1,31 @@
+from typing import Any, Protocol
+
+
+class Task(Protocol):
+    def run(self) -> None:
+        """Run the task."""
+
+
+class TaskFactory(Protocol):
+    def create(self, args: dict[str, Any]) -> Task:
+        """Creates a new task."""
+
+
+class TaskRegistry:
+    def __init__(self):
+        self.registry: dict[str, TaskFactory] = {}
+
+    def register(self, task_type: str, factory: TaskFactory) -> None:
+        self.registry[task_type] = factory
+
+    def unregister(self, task_type: str) -> None:
+        self.registry.pop(task_type, None)
+
+    def create(self, args: dict[str, Any]) -> Task:
+        args_copy = args.copy()
+        task_type = args_copy.pop("type")
+        try:
+            factory = self.registry[task_type]
+        except KeyError:
+            raise ValueError(f"Unknown task type: {task_type!r}") from None
+        return factory.create(args_copy)

pythonic-patterns/registry_pattern/class_based/tasks.json

@@ -0,0 +1,17 @@
+{
+  "tasks": [
+    {
+      "type": "pulse",
+      "strength": 190
+    },
+    {
+      "type": "recalibrate",
+      "target": "Thoron subspace transponder"
+    },
+    {
+      "type": "reinforce",
+      "plating_type": "biogenic",
+      "target": "the deflector array"
+    }
+  ]
+}

pythonic-patterns/registry_pattern/class_based/tasks.py

@@ -0,0 +1,46 @@
+from dataclasses import dataclass
+from typing import Any
+
+from registry import Task
+
+
+@dataclass
+class Pulse:
+    strength: int
+
+    def run(self) -> None:
+        print(
+            f"Sending a subspace pulse of {self.strength} microPicards to the converter assembly."
+        )
+
+
+@dataclass
+class Recalibrate:
+    target: str
+
+    def run(self) -> None:
+        print(f"Recalibrating the {self.target}.")
+
+
+@dataclass
+class Reinforce:
+    plating_type: str
+    target: str
+
+    def run(self) -> None:
+        print(f"Reinforcing {self.plating_type} plating of {self.target}.")
+
+
+class PulseFactory:
+    def create(self, args: dict[str, Any]) -> Task:
+        return Pulse(**args)
+
+
+class RecalibrateFactory:
+    def create(self, args: dict[str, Any]) -> Task:
+        return Recalibrate(**args)
+
+
+class ReinforceFactory:
+    def create(self, args: dict[str, Any]) -> Task:
+        return Reinforce(**args)

pythonic-patterns/registry_pattern/fn_based/inject.py

@@ -0,0 +1,8 @@
+from registry import register
+
+
+def inject(material: str, target: str) -> None:
+    print(f"Injecting {material} into {target}.")
+
+
+register("inject", inject)

pythonic-patterns/registry_pattern/fn_based/loader.py

@@ -0,0 +1,6 @@
+import importlib
+
+
+def load_plugins(plugins: list[str]) -> None:
+    for plugin in plugins:
+        importlib.import_module(plugin)

pythonic-patterns/registry_pattern/fn_based/main.py

@@ -0,0 +1,41 @@
+import json
+
+from loader import load_plugins
+from registry import register, run
+
+
+def send_pulse(strength: int) -> None:
+    print(
+        f"Sending a subspace pulse of {strength} microPicards to the converter assembly."
+    )
+
+
+def recalibrate(target: str) -> None:
+    print(f"Recalibrating the {target}.")
+
+
+def reinforce(plating_type: str, target: str) -> None:
+    print(f"Reinforcing {plating_type} plating of {target}.")
+
+
+def main() -> None:
+
+    # register a couple of tasks
+    register("pulse", send_pulse)
+    register("recalibrate", recalibrate)
+    register("reinforce", reinforce)
+
+    # read data from a JSON file
+    with open("./tasks.json", encoding="utf-8") as file:
+        data = json.load(file)
+
+        # load the plugins
+        load_plugins(data["plugins"])
+
+        # run the tasks
+        for task in data["tasks"]:
+            run(task)
+
+
+if __name__ == "__main__":
+    main()

pythonic-patterns/registry_pattern/fn_based/registry.py

@@ -0,0 +1,17 @@
+from typing import Any, Callable
+
+task_functions: dict[str, Callable[..., None]] = {}
+
+
+def register(task_type: str, task_fn: Callable[..., None]) -> None:
+    task_functions[task_type] = task_fn
+
+
+def unregister(task_type: str) -> None:
+    task_functions.pop(task_type, None)
+
+
+def run(arguments: dict[str, Any]) -> None:
+    args_copy = arguments.copy()
+    task_type = args_copy.pop("type")
+    task_functions[task_type](**args_copy)

pythonic-patterns/registry_pattern/fn_based/tasks.json

@@ -0,0 +1,23 @@
+{
+  "plugins": ["inject"],
+  "tasks": [
+    {
+      "type": "pulse",
+      "strength": 190
+    },
+    {
+      "type": "recalibrate",
+      "target": "Thoron subspace transponder"
+    },
+    {
+      "type": "reinforce",
+      "plating_type": "biogenic",
+      "target": "the deflector array"
+    },
+    {
+      "type": "inject",
+      "material": "tachyons",
+      "target": "molecular transporter resonator"
+    }
+  ]
+}

pythonic-patterns/registry_pattern/readme.md

@@ -0,0 +1,4 @@
+# Registry Pattern
+
+### Summary
+- Video Link: https://www.arjancodes.com/products/the-software-designer-mindset-pythonic-patterns/categories/2149946555/posts/2160000778

pythonic-patterns/registry_pattern/registry_reference/registry_diagram.md

@@ -0,0 +1,71 @@
+```mermaid
+classDiagram
+
+    class AbstractFactory {
+        <<abstract>>
+        Product create()*
+    }
+    class ConcreteFactory1 {
+        Product create()
+    }
+    class ConcreteFactory2 {
+        Product create()
+    }
+    class Product {
+        <<abstract>>
+    }
+    class Registry {
+        registry: dict[str, AbstractFactory]
+        register(type: str, factory: AbstractFactory)
+        unregister(type: str)
+        Product create(type: str)
+    }
+
+    ConcreteFactory1 --|> AbstractFactory
+    ConcreteFactory2 --|> AbstractFactory
+    ConcreteProduct1 --|> Product
+    ConcreteProduct2 --|> Product
+    ConcreteFactory1 ..> ConcreteProduct1
+    ConcreteFactory2 ..> ConcreteProduct2
+    Registry o-- AbstractFactory
+```
+
+```mermaid
+classDiagram
+
+    class TaskFactory {
+        <<abstract>>
+        Task create(args)*
+    }
+    class PulseFactory {
+        Task create(args)
+    }
+    class RecalibrateFactory {
+        Task create(args)
+    }
+    class ReinforceFactory {
+        Task create(args)
+    }
+    class Task {
+        <<abstract>>
+        run()*
+    }
+    class TaskRegistry {
+        registry: dict[str, TaskFactory]
+        register(type: str, factory: TaskFactory)
+        unregister(type: str)
+        Task create(type: str)
+    }
+
+    PulseFactory --|> TaskFactory
+    RecalibrateFactory --|> TaskFactory
+    ReinforceFactory --|> TaskFactory
+    Pulse --|> Task
+    Recalibrate --|> Task
+    Reinforce --|> Task
+
+    PulseFactory ..> Pulse
+    RecalibrateFactory ..> Recalibrate
+    ReinforceFactory ..> Reinforce
+    TaskRegistry o-- TaskFactory
+```