[WEBSOCKET]: Channels

jac-cloud/jac_cloud/jaseci/__init__.py

@@ -62,7 +62,7 @@ async def lifespan(app: _FaststAPI) -> AsyncGenerator[None, _FaststAPI]:
             populate_yaml_specs(cls.__app__)
 
             from .routers import healthz_router, sso_router, user_router
-            from ..plugin.jaseci import walker_router, websocket_router
+            from ..plugin.implementation import walker_router, websocket_router
 
             for router in [
                 healthz_router,

jac-cloud/jac_cloud/jaseci/dtos/__init__.py

@@ -8,6 +8,13 @@
     UserResetPassword,
     UserVerification,
 )
+from .websocket import (
+    ChannelEvent,
+    ConnectionEvent,
+    UserEvent,
+    WalkerEvent,
+    WebSocketEvent,
+)
 
 
 __all__ = [
@@ -18,4 +25,9 @@
     "UserRequest",
     "UserResetPassword",
     "UserVerification",
+    "ChannelEvent",
+    "ConnectionEvent",
+    "UserEvent",
+    "WalkerEvent",
+    "WebSocketEvent",
 ]

jac-cloud/jac_cloud/jaseci/dtos/websocket.py

@@ -0,0 +1,43 @@
+"""Jaseci SSO DTOs."""
+
+from typing import Annotated, Literal, Union
+
+from pydantic import BaseModel, Field
+
+
+class ConnectionEvent(BaseModel):
+    """Connection Event Model."""
+
+    type: Literal["connection"]
+
+
+class WalkerEvent(BaseModel):
+    """Walker Event Model."""
+
+    type: Literal["walker"]
+    data: dict
+
+
+class UserEvent(BaseModel):
+    """Walker Event Model."""
+
+    type: Literal["user"]
+    root_id: str
+    data: dict
+
+
+class ChannelEvent(BaseModel):
+    """Walker Event Model."""
+
+    type: Literal["channel"]
+    channel_id: str
+    data: dict
+
+
+class WebSocketEvent(BaseModel):
+    """WebSocket Event."""
+
+    event: Annotated[
+        Union[ConnectionEvent, WalkerEvent, UserEvent, ChannelEvent],
+        Field(discriminator="type"),
+    ]

jac-cloud/jac_cloud/plugin/__init__.py

@@ -1,6 +1,6 @@
 """Jaseci Plugins."""
 
-from .jaseci import specs
+from .implementation import WEBSOCKET_MANAGER, specs
 
 
-__all__ = ["specs"]
+__all__ = ["WEBSOCKET_MANAGER", "specs"]

jac-cloud/jac_cloud/plugin/implementation/__init__.py

@@ -0,0 +1,6 @@
+"""Jaseci Plugin Implementations."""
+
+from .api import specs, walker_router
+from .websocket import WEBSOCKET_MANAGER, websocket_router
+
+__all__ = ["specs", "walker_router", "WEBSOCKET_MANAGER", "websocket_router"]

jac-cloud/jac_cloud/plugin/implementation/api.py

@@ -0,0 +1,302 @@
+"""Jaseci API Implementations."""
+
+from dataclasses import Field, MISSING, fields, is_dataclass
+from os import getenv
+from re import compile
+from types import NoneType
+from typing import Any, Callable, Type, TypeAlias, Union, cast, get_type_hints
+
+from asyncer import syncify
+
+from fastapi import (
+    APIRouter,
+    Depends,
+    File,
+    HTTPException,
+    Request,
+    Response,
+    UploadFile,
+)
+from fastapi.responses import ORJSONResponse
+
+from jaclang.plugin.feature import JacFeature as Jac
+
+from orjson import loads
+
+from pydantic import BaseModel, Field as pyField, ValidationError, create_model
+
+from starlette.datastructures import UploadFile as BaseUploadFile
+
+from .websocket import websocket_events
+from ...core.architype import (
+    Anchor,
+    NodeAnchor,
+    WalkerAnchor,
+    WalkerArchitype,
+)
+from ...core.context import ContextResponse, JaseciContext
+from ...jaseci.security import authenticator
+
+# from ..jaseci.utils import log_entry, log_exit
+
+DISABLE_AUTO_ENDPOINT = getenv("DISABLE_AUTO_ENDPOINT") == "true"
+PATH_VARIABLE_REGEX = compile(r"{([^\}]+)}")
+FILE_TYPES = {
+    UploadFile,
+    list[UploadFile],
+    UploadFile | None,
+    list[UploadFile] | None,
+}
+
+walker_router = APIRouter(prefix="/walker", tags=["walker"])
+
+
+class DefaultSpecs:
+    """Default API specs."""
+
+    path: str = ""
+    methods: list[str] = ["post"]
+    as_query: str | list[str] = []
+    excluded: str | list[str] = []
+    auth: bool = True
+    private: bool = False
+
+
+def get_specs(cls: type) -> Type["DefaultSpecs"] | None:
+    """Get Specs and inherit from DefaultSpecs."""
+    specs = getattr(cls, "__specs__", None)
+    if specs is None:
+        if DISABLE_AUTO_ENDPOINT:
+            return None
+        specs = DefaultSpecs
+
+    if not issubclass(specs, DefaultSpecs):
+        specs = type(specs.__name__, (specs, DefaultSpecs), {})
+
+    return specs
+
+
+def gen_model_field(cls: type, field: Field, is_file: bool = False) -> tuple[type, Any]:
+    """Generate Specs for Model Field."""
+    if field.default is not MISSING:
+        consts = (cls, pyField(default=field.default))
+    elif callable(field.default_factory):
+        consts = (cls, pyField(default_factory=field.default_factory))
+    else:
+        consts = (cls, File(...) if is_file else ...)
+
+    return consts
+
+
+def populate_apis(cls: Type[WalkerArchitype]) -> None:
+    """Generate FastAPI endpoint based on WalkerArchitype class."""
+    if (specs := get_specs(cls)) and not specs.private:
+        path: str = specs.path or ""
+        methods: list = specs.methods or []
+        as_query: str | list[str] = specs.as_query or []
+        excluded: str | list[str] = specs.excluded or []
+        auth: bool = specs.auth or False
+
+        query: dict[str, Any] = {}
+        body: dict[str, Any] = {}
+        files: dict[str, Any] = {}
+        message: dict[str, Any] = {}
+
+        if path:
+            if not path.startswith("/"):
+                path = f"/{path}"
+            if isinstance(as_query, list):
+                as_query += PATH_VARIABLE_REGEX.findall(path)
+
+        hintings = get_type_hints(cls)
+
+        if is_dataclass(cls):
+            if excluded != "*":
+                if isinstance(excluded, str):
+                    excluded = [excluded]
+
+                for f in fields(cls):
+                    if f.name in excluded:
+                        if f.default is MISSING and not callable(f.default_factory):
+                            raise AttributeError(
+                                f"{cls.__name__} {f.name} should have default or default_factory."
+                            )
+                        continue
+
+                    f_name = f.name
+                    f_type = hintings[f_name]
+                    if f_type in FILE_TYPES:
+                        message[f_name] = files[f_name] = gen_model_field(
+                            f_type, f, True
+                        )
+                    else:
+                        consts = gen_model_field(f_type, f)
+                        message[f_name] = consts
+
+                        if as_query == "*" or f_name in as_query:
+                            query[f_name] = consts
+                        else:
+                            body[f_name] = consts
+            elif any(
+                f.default is MISSING and not callable(f.default_factory)
+                for f in fields(cls)
+            ):
+                raise AttributeError(
+                    f"{cls.__name__} fields should all have default or default_factory."
+                )
+
+        payload: dict[str, Any] = {
+            "query": (
+                create_model(f"{cls.__name__.lower()}_query_model", **query),
+                Depends(),
+            ),
+            "files": (
+                create_model(f"{cls.__name__.lower()}_files_model", **files),
+                Depends(),
+            ),
+        }
+
+        body_model = None
+        if body:
+            body_model = create_model(f"{cls.__name__.lower()}_body_model", **body)
+
+            if files:
+                payload["body"] = (UploadFile, File(...))
+            else:
+                payload["body"] = (body_model, ...)
+
+        payload_model = create_model(f"{cls.__name__.lower()}_request_model", **payload)
+
+        def api_entry(
+            request: Request,
+            node: str | None,
+            payload: payload_model = Depends(),  # type: ignore # noqa: B008
+        ) -> ORJSONResponse:
+            pl = cast(BaseModel, payload).model_dump()
+            body = pl.get("body", {})
+
+            # log = log_entry(
+            #     cls.__name__,
+            #     user.email if (user := getattr(request, "_user", None)) else None,
+            #     pl,
+            #     node,
+            # )
+
+            if isinstance(body, BaseUploadFile) and body_model:
+                body = loads(syncify(body.read)())
+                try:
+                    body = body_model(**body).__dict__
+                except ValidationError as e:
+                    return ORJSONResponse({"detail": e.errors()})
+
+            jctx = JaseciContext.create(request, NodeAnchor.ref(node) if node else None)
+
+            wlk: WalkerAnchor = cls(**body, **pl["query"], **pl["files"]).__jac__
+            if Jac.check_read_access(jctx.entry_node):
+                Jac.spawn_call(wlk.architype, jctx.entry_node.architype)
+                jctx.close()
+
+                if jctx.custom is not MISSING:
+                    return jctx.custom
+
+                resp = jctx.response(wlk.returns)
+                # log_exit(resp, log)
+
+                return ORJSONResponse(resp, jctx.status)
+            else:
+                error = {
+                    "error": f"You don't have access on target entry{cast(Anchor, jctx.entry_node).ref_id}!"
+                }
+                jctx.close()
+
+                # log_exit(error, log)
+                raise HTTPException(403, error)
+
+        def api_root(
+            request: Request,
+            payload: payload_model = Depends(),  # type: ignore # noqa: B008
+        ) -> Response:
+            return api_entry(request, None, payload)
+
+        for method in methods:
+            method = method.lower()
+            walker_method = getattr(walker_router, method)
+
+            match method:
+                case "websocket":
+                    websocket_events[cls.__name__] = {
+                        "type": cls,
+                        "model": create_model(
+                            f"{cls.__name__.lower()}_message_model", **message
+                        ),
+                        "auth": auth,
+                    }
+                case _:
+                    raw_types: list[Type] = [
+                        get_type_hints(jef.func).get("return", NoneType)
+                        for jef in (*cls._jac_entry_funcs_, *cls._jac_exit_funcs_)
+                    ]
+
+                    if raw_types:
+                        if len(raw_types) > 1:
+                            ret_types: TypeAlias = Union[*raw_types]  # type: ignore[valid-type]
+                        else:
+                            ret_types = raw_types[0]  # type: ignore[misc]
+                    else:
+                        ret_types = NoneType  # type: ignore[misc]
+
+                    settings: dict[str, Any] = {
+                        "tags": ["walker"],
+                        "response_model": ContextResponse[ret_types] | Any,
+                    }
+                    if auth:
+                        settings["dependencies"] = cast(list, authenticator)
+
+                    walker_method(
+                        url := f"/{cls.__name__}{path}", summary=url, **settings
+                    )(api_root)
+                    walker_method(
+                        url := f"/{cls.__name__}/{{node}}{path}",
+                        summary=url,
+                        **settings,
+                    )(api_entry)
+
+
+def specs(
+    cls: Type[WalkerArchitype] | None = None,
+    *,
+    path: str = "",
+    methods: list[str] = ["post"],  # noqa: B006
+    as_query: str | list[str] = [],  # noqa: B006
+    excluded: str | list[str] = [],  # noqa: B006
+    auth: bool = True,
+    private: bool = False,
+) -> Callable:
+    """Walker Decorator."""
+
+    def wrapper(cls: Type[WalkerArchitype]) -> Type[WalkerArchitype]:
+        if get_specs(cls) is None:
+            p = path
+            m = methods
+            aq = as_query
+            ex = excluded
+            a = auth
+            pv = private
+
+            class __specs__(DefaultSpecs):  # noqa: N801
+                path: str = p
+                methods: list[str] = m
+                as_query: str | list[str] = aq
+                excluded: str | list[str] = ex
+                auth: bool = a
+                private: bool = pv
+
+            cls.__specs__ = __specs__  # type: ignore[attr-defined]
+
+            populate_apis(cls)
+        return cls
+
+    if cls:
+        return wrapper(cls)
+
+    return wrapper