[POC]: Optional Patching

Author: amadolid

jaclang_jaseci/jaseci/__init__.py

@@ -15,6 +15,33 @@ class FastAPI:
     """FastAPI Handler."""
 
     __app__ = None
+    __is_imported__: bool | None = None
+
+    @staticmethod
+    def is_imported() -> bool:
+        """Check if fastapi is used."""
+        from jaclang.plugin.feature import JacFeature as Jac
+        from jaclang.runtimelib.machine import JacMachine
+
+        from ..core.architype import (
+            EdgeArchitype,
+            NodeArchitype,
+            ObjectArchitype,
+            Root,
+            WalkerArchitype,
+        )
+
+        if not isinstance(FastAPI.__is_imported__, bool):
+            main = JacMachine.get().loaded_modules.get("__main__")
+            FastAPI.__is_imported__ = getattr(main, "FastAPI", None) is FastAPI
+            if FastAPI.__is_imported__:
+                Jac.RootType = Root  # type: ignore[assignment]
+                Jac.Obj = ObjectArchitype  # type: ignore[assignment]
+                Jac.Node = NodeArchitype  # type: ignore[assignment]
+                Jac.Edge = EdgeArchitype  # type: ignore[assignment]
+                Jac.Walker = WalkerArchitype  # type: ignore[assignment]
+
+        return FastAPI.__is_imported__
 
     @classmethod
     def get(cls) -> _FaststAPI:

jaclang_jaseci/plugin/jaseci.py

@@ -20,7 +20,7 @@
 )
 from fastapi.responses import ORJSONResponse
 
-from jaclang.plugin.default import hookimpl
+from jaclang.plugin.default import JacFeatureDefaults, hookimpl
 from jaclang.plugin.feature import JacFeature as Jac
 from jaclang.runtimelib.architype import DSFunc
 
@@ -42,7 +42,8 @@
     WalkerAnchor,
     WalkerArchitype,
 )
-from ..core.context import JaseciContext
+from ..core.context import ExecutionContext, JaseciContext
+from ..jaseci import FastAPI
 from ..jaseci.security import authenticator
 
 
@@ -243,9 +244,11 @@ class JacPlugin:
 
     @staticmethod
     @hookimpl
-    def get_context() -> JaseciContext:
+    def get_context() -> ExecutionContext:
         """Get current execution context."""
-        return JaseciContext.get()
+        if FastAPI.is_imported():
+            return JaseciContext.get()
+        return JacFeatureDefaults.get_context()
 
     @staticmethod
     @hookimpl
@@ -256,125 +259,150 @@ def make_architype(
         on_exit: list[DSFunc],
     ) -> Type[Architype]:
         """Create a new architype."""
-        for i in on_entry + on_exit:
-            i.resolve(cls)
-        if not hasattr(cls, "_jac_entry_funcs_") or not hasattr(
-            cls, "_jac_exit_funcs_"
-        ):
-            # Saving the module path and reassign it after creating cls
-            # So the jac modules are part of the correct module
-            cur_module = cls.__module__
-            cls = type(cls.__name__, (cls, arch_base), {})
-            cls.__module__ = cur_module
-            cls._jac_entry_funcs_ = on_entry  # type: ignore
-            cls._jac_exit_funcs_ = on_exit  # type: ignore
-        else:
-            new_entry_funcs = OrderedDict(zip([i.name for i in on_entry], on_entry))
-            entry_funcs = OrderedDict(
-                zip([i.name for i in cls._jac_entry_funcs_], cls._jac_entry_funcs_)
-            )
-            entry_funcs.update(new_entry_funcs)
-            cls._jac_entry_funcs_ = list(entry_funcs.values())
+        if FastAPI.is_imported():
+            for i in on_entry + on_exit:
+                i.resolve(cls)
+            if not hasattr(cls, "_jac_entry_funcs_") or not hasattr(
+                cls, "_jac_exit_funcs_"
+            ):
+                # Saving the module path and reassign it after creating cls
+                # So the jac modules are part of the correct module
+                cur_module = cls.__module__
+                cls = type(cls.__name__, (cls, arch_base), {})
+                cls.__module__ = cur_module
+                cls._jac_entry_funcs_ = on_entry  # type: ignore
+                cls._jac_exit_funcs_ = on_exit  # type: ignore
+            else:
+                new_entry_funcs = OrderedDict(zip([i.name for i in on_entry], on_entry))
+                entry_funcs = OrderedDict(
+                    zip([i.name for i in cls._jac_entry_funcs_], cls._jac_entry_funcs_)
+                )
+                entry_funcs.update(new_entry_funcs)
+                cls._jac_entry_funcs_ = list(entry_funcs.values())
 
-            new_exit_funcs = OrderedDict(zip([i.name for i in on_exit], on_exit))
-            exit_funcs = OrderedDict(
-                zip([i.name for i in cls._jac_exit_funcs_], cls._jac_exit_funcs_)
-            )
-            exit_funcs.update(new_exit_funcs)
-            cls._jac_exit_funcs_ = list(exit_funcs.values())
+                new_exit_funcs = OrderedDict(zip([i.name for i in on_exit], on_exit))
+                exit_funcs = OrderedDict(
+                    zip([i.name for i in cls._jac_exit_funcs_], cls._jac_exit_funcs_)
+                )
+                exit_funcs.update(new_exit_funcs)
+                cls._jac_exit_funcs_ = list(exit_funcs.values())
 
-        inner_init = cls.__init__  # type: ignore
+            inner_init = cls.__init__  # type: ignore
 
-        @wraps(inner_init)
-        def new_init(self: Architype, *args: object, **kwargs: object) -> None:
-            arch_base.__init__(self)
-            inner_init(self, *args, **kwargs)
+            @wraps(inner_init)
+            def new_init(self: Architype, *args: object, **kwargs: object) -> None:
+                arch_base.__init__(self)
+                inner_init(self, *args, **kwargs)
 
-        cls.__init__ = new_init  # type: ignore
-        return cls
+            cls.__init__ = new_init  # type: ignore
+            return cls
+        return JacFeatureDefaults.make_architype(
+            cls=cls, arch_base=arch_base, on_entry=on_entry, on_exit=on_exit
+        )
 
     @staticmethod
     @hookimpl
     def make_obj(
         on_entry: list[DSFunc], on_exit: list[DSFunc]
     ) -> Callable[[type], type]:
         """Create a new architype."""
+        if FastAPI.is_imported():
+
+            def decorator(cls: Type[Architype]) -> Type[Architype]:
+                """Decorate class."""
+                cls = Jac.make_architype(
+                    cls=cls,
+                    arch_base=ObjectArchitype,
+                    on_entry=on_entry,
+                    on_exit=on_exit,
+                )
+                return cls
 
-        def decorator(cls: Type[Architype]) -> Type[Architype]:
-            """Decorate class."""
-            cls = Jac.make_architype(
-                cls=cls, arch_base=ObjectArchitype, on_entry=on_entry, on_exit=on_exit
-            )
-            return cls
-
-        return decorator
+            return decorator
+        return JacFeatureDefaults.make_obj(on_entry=on_entry, on_exit=on_exit)
 
     @staticmethod
     @hookimpl
     def make_node(
         on_entry: list[DSFunc], on_exit: list[DSFunc]
     ) -> Callable[[type], type]:
         """Create a obj architype."""
+        if FastAPI.is_imported():
 
-        def decorator(cls: Type[Architype]) -> Type[Architype]:
-            """Decorate class."""
-            cls = Jac.make_architype(
-                cls=cls, arch_base=NodeArchitype, on_entry=on_entry, on_exit=on_exit
-            )
-            return cls
+            def decorator(cls: Type[Architype]) -> Type[Architype]:
+                """Decorate class."""
+                cls = Jac.make_architype(
+                    cls=cls, arch_base=NodeArchitype, on_entry=on_entry, on_exit=on_exit
+                )
+                return cls
 
-        return decorator
+            return decorator
+        return JacFeatureDefaults.make_node(on_entry=on_entry, on_exit=on_exit)
 
     @staticmethod
     @hookimpl
     def make_edge(
         on_entry: list[DSFunc], on_exit: list[DSFunc]
     ) -> Callable[[type], type]:
         """Create a edge architype."""
+        if FastAPI.is_imported():
 
-        def decorator(cls: Type[Architype]) -> Type[Architype]:
-            """Decorate class."""
-            cls = Jac.make_architype(
-                cls=cls, arch_base=EdgeArchitype, on_entry=on_entry, on_exit=on_exit
-            )
-            return cls
+            def decorator(cls: Type[Architype]) -> Type[Architype]:
+                """Decorate class."""
+                cls = Jac.make_architype(
+                    cls=cls, arch_base=EdgeArchitype, on_entry=on_entry, on_exit=on_exit
+                )
+                return cls
 
-        return decorator
+            return decorator
+        return JacFeatureDefaults.make_edge(on_entry=on_entry, on_exit=on_exit)
 
     @staticmethod
     @hookimpl
     def make_walker(
         on_entry: list[DSFunc], on_exit: list[DSFunc]
     ) -> Callable[[type], type]:
         """Create a walker architype."""
+        if FastAPI.is_imported():
+
+            def decorator(cls: Type[Architype]) -> Type[Architype]:
+                """Decorate class."""
+                cls = Jac.make_architype(
+                    cls=cls,
+                    arch_base=WalkerArchitype,
+                    on_entry=on_entry,
+                    on_exit=on_exit,
+                )
+                populate_apis(cls)
+                return cls
 
-        def decorator(cls: Type[Architype]) -> Type[Architype]:
-            """Decorate class."""
-            cls = Jac.make_architype(
-                cls=cls, arch_base=WalkerArchitype, on_entry=on_entry, on_exit=on_exit
-            )
-            populate_apis(cls)
-            return cls
-
-        return decorator
+            return decorator
+        return JacFeatureDefaults.make_walker(on_entry=on_entry, on_exit=on_exit)
 
     @staticmethod
     @hookimpl
-    def report(expr: Any) -> Any:  # noqa: ANN401
+    def report(expr: Any) -> None:  # noqa:ANN401
         """Jac's report stmt feature."""
-        JaseciContext.get().reports.append(expr)
+        if FastAPI.is_imported():
+            JaseciContext.get().reports.append(expr)
+            return
+        JacFeatureDefaults.report(expr=expr)
 
     @staticmethod
     @hookimpl
     def get_root() -> Root:
         """Jac's assign comprehension feature."""
-        return JaseciContext.get_root()
+        if FastAPI.is_imported():
+            return JaseciContext.get_root()
+        return JacFeatureDefaults.get_root()  # type:ignore[return-value]
 
     @staticmethod
     @hookimpl
     def get_root_type() -> Type[Root]:
         """Jac's root getter."""
-        return Root
+        if FastAPI.is_imported():
+            return Root
+        return JacFeatureDefaults.get_root_type()  # type:ignore[return-value]
 
     @staticmethod
     @hookimpl
@@ -384,32 +412,25 @@ def build_edge(
         conn_assign: tuple[tuple, tuple] | None,
     ) -> Callable[[NodeAnchor, NodeAnchor], EdgeArchitype]:
         """Jac's root getter."""
-        conn_type = conn_type if conn_type else GenericEdge
-
-        def builder(source: NodeAnchor, target: NodeAnchor) -> EdgeArchitype:
-            edge = conn_type() if isinstance(conn_type, type) else conn_type
-            edge.__attach__(source, target, is_undirected)
-            if conn_assign:
-                for fld, val in zip(conn_assign[0], conn_assign[1]):
-                    if hasattr(edge, fld):
-                        setattr(edge, fld, val)
-                    else:
-                        raise ValueError(f"Invalid attribute: {fld}")
-            if source.persistent or target.persistent:
-                edge.__jac__.save()
-                target.save()
-                source.save()
-            return edge
-
-        return builder
-
-
-##########################################################
-#               NEED TO TRANSFER TO PLUGIN               #
-##########################################################
-
-Jac.RootType = Root  # type: ignore[assignment]
-Jac.Obj = ObjectArchitype  # type: ignore[assignment]
-Jac.Node = NodeArchitype  # type: ignore[assignment]
-Jac.Edge = EdgeArchitype  # type: ignore[assignment]
-Jac.Walker = WalkerArchitype  # type: ignore[assignment]
+        if FastAPI.is_imported():
+            conn_type = conn_type if conn_type else GenericEdge
+
+            def builder(source: NodeAnchor, target: NodeAnchor) -> EdgeArchitype:
+                edge = conn_type() if isinstance(conn_type, type) else conn_type
+                edge.__attach__(source, target, is_undirected)
+                if conn_assign:
+                    for fld, val in zip(conn_assign[0], conn_assign[1]):
+                        if hasattr(edge, fld):
+                            setattr(edge, fld, val)
+                        else:
+                            raise ValueError(f"Invalid attribute: {fld}")
+                if source.persistent or target.persistent:
+                    edge.__jac__.save()
+                    target.save()
+                    source.save()
+                return edge
+
+            return builder
+        return JacFeatureDefaults.build_edge(  # type:ignore[return-value]
+            is_undirected=is_undirected, conn_type=conn_type, conn_assign=conn_assign
+        )