upload pyapp

pyapp/stereotype.py

@@ -0,0 +1,106 @@
+import inspect
+
+from pyapp.component import Component
+from pyapp.utils.log import Log
+
+
+def ensure_component_subclass(cls, type_name, name):
+    if not issubclass(cls, Component):
+        new_cls  = type(cls.__name__, (Component, cls), {
+            **cls.__dict__,
+            "TYPE": type_name,
+            "NAME": name
+        })
+        new_cls.__original_class__ = cls
+    else:
+        new_cls = cls
+    for method_name, method in inspect.getmembers(cls, predicate=inspect.isfunction):
+        if getattr(method, "__isabstractmethod__", False):
+            Log.error(f"Component <{name}> contains abstract method <{method_name}>.", True)
+        
+    return cls
+
+
+all_component_classes = {}
+all_component_comments = {}
+
+def register_component_type(component_type, comment=None):
+    all_component_classes[component_type] = {}
+    all_component_comments[component_type] = comment
+
+# --- Classes --- #
+
+def component(component_type, component_name, comment=None):
+    def decorator(cls):
+        cls = ensure_component_subclass(cls, "COMPONENT", component_name)
+        if component_type not in all_component_classes:
+            register_component_type(component_type, comment)
+        all_component_classes[component_type][component_name] = cls
+        all_component_comments[component_type][component_name] = comment
+        return cls
+    return decorator
+
+runner_classes = {}
+runner_comments = {}
+def runner(runner_name, comment=None):
+    def decorator(cls):
+        if not hasattr(cls, 'run') or not callable(getattr(cls, 'run')):
+            Log.error(f"runner <{cls.__name__}> must implement a 'run' method", True)
+        cls = ensure_component_subclass(cls, "RUNNER", runner_name)
+        runner_classes[runner_name] = cls
+        runner_comments[runner_name] = comment
+        return cls
+    return decorator
+
+factory_classes = {}
+factory_comments = {}
+def factory(factory_name, comment=None):
+    def decorator(cls):
+        if not hasattr(cls, 'create') or not callable(getattr(cls, 'create')):
+            Log.error(f"factory <{cls.__name__}> must implement a 'create' method", True)
+        
+        cls = ensure_component_subclass(cls, "FACTORY", factory_name)
+        factory_classes[factory_name] = cls
+        factory_comments[factory_name] = comment
+        return cls
+    return decorator
+
+client_classes = {}
+client_comments = {}
+def client(client_name, comment=None):
+    def decorator(cls):
+        cls = ensure_component_subclass(cls, "CLIENT", client_name)
+        client_classes[client_name] = cls
+        client_comments[client_name] = comment
+        return cls
+    return decorator
+
+server_classes = {}
+server_comments = {}
+def server(server_name, comment=None):
+    def decorator(cls):
+        cls = ensure_component_subclass(cls, "SERVER", server_name)
+        if not hasattr(cls, 'host') or not hasattr(cls, 'port'):
+            Log.error(f"server <{cls.__name__}> must implement 'host' and 'port' attributes", True)
+        if not hasattr(cls, 'serve') or not callable(getattr(cls, 'serve')):
+            Log.error(f"server <{cls.__name__}> must implement a 'serve' method", True)
+        server_classes[server_name] = cls
+        server_comments[server_name] = comment
+        return cls
+    return decorator
+
+# --- Utils --- #
+
+register_component_type("RUNNER", "Runner")
+register_component_type("FACTORY", "Factory")
+register_component_type("CLIENT", "Client")
+register_component_type("SERVER", "Server")
+
+def get_all_component_classes():
+    return all_component_classes
+
+def get_all_component_comments():
+    return all_component_comments
+
+def get_component_classes(component_type):
+    return all_component_classes.get(component_type, None)