temporary

jac/jaclang/runtimelib/context.py

@@ -14,9 +14,7 @@
 
 EXECUTION_CONTEXT = ContextVar[Optional["ExecutionContext"]]("ExecutionContext")
 
-SUPER_ROOT_JID = JID[NodeAnchor](
- id=UUID("00000000-0000-0000-0000-000000000000"), type=NodeAnchor
-)
+SUPER_ROOT_JID = JID(id=UUID("00000000-0000-0000-0000-000000000000"), type=NodeAnchor)
 SUPER_ROOT_ARCHITYPE = object.__new__(Root)
 SUPER_ROOT_ANCHOR = NodeAnchor(
  jid=SUPER_ROOT_JID, architype=SUPER_ROOT_ARCHITYPE, persistent=False, edge_ids=set()

jac/jaclang/runtimelib/implementation.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass, field
 from logging import getLogger
 from re import IGNORECASE, compile
-from typing import Type, TypeAlias
+from typing import Type, TypeAlias, TypeVar
 from uuid import UUID, uuid4
 
 from .interface import (
@@ -17,7 +17,6 @@
  Permission,
  WalkerAnchor as _WalkerAnchor,
  WalkerArchitype as _WalkerArchitype,
- _ANCHOR,
 )
 
 
@@ -27,22 +26,24 @@
 )
 
 
+_ANCHOR = TypeVar("_ANCHOR", "NodeAnchor", "EdgeAnchor", "WalkerAnchor", covariant=True)
 Anchor: TypeAlias = "NodeAnchor" | "EdgeAnchor" | "WalkerAnchor"
 Architype: TypeAlias = "NodeArchitype" | "EdgeArchitype" | "WalkerArchitype"
 logger = getLogger(__name__)
 
 
 @dataclass(kw_only=True)
-class JID(_JID[UUID, _ANCHOR]):
+class JID(_JID[UUID, Anchor]):
  """Jaclang Default JID."""
 
  id: UUID
+ type: Type[Anchor]
  name: str
 
  def __init__(
- self: JID[Anchor],
+ self,
  id: str | UUID | None = None,
- type: Type[Anchor] | None = None,
+ type: Type[_ANCHOR] | None = None,
  name: str = "",
  ) -> None:
  """Override JID initializer."""
@@ -85,7 +86,7 @@ def __str__(self) -> str:
 class NodeAnchor(_NodeAnchor["NodeAnchor"]):
  """NodeAnchor Interface."""
 
- jid: JID[NodeAnchor] = field(default_factory=lambda: JID(type=NodeAnchor))
+ jid: JID["NodeAnchor"] = field(default_factory=lambda: JID(type=NodeAnchor))
  architype: "NodeArchitype"
  root: JID["NodeAnchor"] | None = None
  access: Permission = field(default_factory=Permission)
@@ -104,6 +105,9 @@ def __deserialize__(cls, data: NodeAnchor) -> NodeAnchor:
  return data
 
 
+aa = JID(id=UUID(), type=NodeAnchor)
+
+
 @dataclass(kw_only=True)
 class EdgeAnchor(_EdgeAnchor["EdgeAnchor"]):
  """NodeAnchor Interface."""

jac/jaclang/runtimelib/interface.py

@@ -6,11 +6,19 @@
 from dataclasses import dataclass, field
 from enum import IntEnum
 from types import UnionType
-from typing import Any, Callable, Generator, Generic, Iterable, Type, TypeVar
+from typing import (
+ Any,
+ Callable,
+ Generator,
+ Generic,
+ Iterable,
+ Type,
+ TypeVar,
+)
 
 
 _ID = TypeVar("_ID")
-_ANCHOR = TypeVar("_ANCHOR", bound="Anchor")
+_ANCHOR = TypeVar("_ANCHOR", "NodeAnchor", "EdgeAnchor", "WalkerAnchor", covariant=True)
 _NODE_ANCHOR = TypeVar("_NODE_ANCHOR", bound="NodeAnchor")
 
 _SERIALIZE = TypeVar("_SERIALIZE")
@@ -75,12 +83,12 @@ class Permission:
 
 
 @dataclass(kw_only=True)
-class Anchor(Generic[_SERIALIZE], ABC):
+class Anchor(Generic[_ID, _SERIALIZE], ABC):
  """Anchor Interface."""
 
- jid: JID
- architype: "Architype"
- root: JID | None
+ jid: JID[_ID, NodeAnchor] | JID[_ID, EdgeAnchor] | JID[_ID, WalkerAnchor]
+ architype: "NodeArchitype" | "EdgeArchitype" | "WalkerArchitype"
+ root: JID[_ID, NodeAnchor] | None
  access: Permission
 
  @abstractmethod
@@ -94,26 +102,29 @@ def __deserialize__(cls: Type[_DESERIALIZE], data: _SERIALIZE) -> _DESERIALIZE:
 
 
 @dataclass(kw_only=True)
-class NodeAnchor(Anchor[_SERIALIZE]):
+class NodeAnchor(Anchor[_ID, _SERIALIZE]):
  """NodeAnchor Interface."""
 
+ jid: JID[_ID, NodeAnchor]
  architype: "NodeArchitype"
- edge_ids: Iterable[JID]
+ edge_ids: Iterable[JID[_ID, EdgeAnchor]]
 
 
 @dataclass(kw_only=True)
-class EdgeAnchor(Anchor[_SERIALIZE]):
+class EdgeAnchor(Anchor[_ID, _SERIALIZE]):
  """EdgeAnchor Interface."""
 
+ jid: JID[_ID, EdgeAnchor]
  architype: "EdgeArchitype"
- source_id: JID
- target_id: JID
+ source_id: JID[_ID, NodeAnchor]
+ target_id: JID[_ID, NodeAnchor]
 
 
 @dataclass(kw_only=True)
-class WalkerAnchor(Anchor[_SERIALIZE]):
+class WalkerAnchor(Anchor[_ID, _SERIALIZE]):
  """WalkerAnchor Interface."""
 
+ jid: JID[_ID, WalkerAnchor]
  architype: "WalkerArchitype"
 
 
@@ -125,7 +136,7 @@ class WalkerAnchor(Anchor[_SERIALIZE]):
 class Architype(Generic[_SERIALIZE], ABC):
  """Architype Interface."""
 
- __jac__: Anchor
+ __jac__: NodeAnchor | EdgeAnchor | WalkerAnchor
 
  @abstractmethod
  def __serialize__(self) -> _SERIALIZE:
@@ -187,8 +198,8 @@ def resolve(self, cls: type) -> None:
 class Memory(Generic[_ID, _ANCHOR]):
  """Generic Memory Handler."""
 
- __mem__: dict[_ID, _ANCHOR] = field(default_factory=dict)
- __gc__: set[_ID] = field(default_factory=set)
+ __mem__: dict[JID[_ID, _ANCHOR], _ANCHOR] = field(default_factory=dict)
+ __gc__: set[JID[_ID, _ANCHOR]] = field(default_factory=set)
 
  def close(self) -> None:
  """Close memory handler."""
@@ -197,36 +208,38 @@ def close(self) -> None:
 
  def find(
  self,
- ids: _ID | Iterable[_ID],
+ ids: JID[_ID, _ANCHOR] | Iterable[JID[_ID, _ANCHOR]],
  filter: Callable[[_ANCHOR], _ANCHOR] | None = None,
  ) -> Generator[_ANCHOR, None, None]:
  """Find anchors from memory by ids with filter."""
  if not isinstance(ids, Iterable):
  ids = [ids]
 
- return (
- anchor
- for id in ids
- if (anchor := self.__mem__.get(id)) and (not filter or filter(anchor))
- )
+ for id in ids:
+ if (
+ (anchor := self.__mem__.get(id))
+ and isinstance(anchor, id.type)
+ and (not filter or filter(anchor))
+ ):
+ yield anchor
 
  def find_one(
  self,
- ids: _ID | Iterable[_ID],
+ ids: JID[_ID, _ANCHOR] | Iterable[JID[_ID, _ANCHOR]],
  filter: Callable[[_ANCHOR], _ANCHOR] | None = None,
  ) -> _ANCHOR | None:
  """Find one anchor from memory by ids with filter."""
  return next(self.find(ids, filter), None)
 
- def find_by_id(self, id: _ID) -> _ANCHOR | None:
+ def find_by_id(self, id: JID[_ID, _ANCHOR]) -> _ANCHOR | None:
  """Find one by id."""
  return self.__mem__.get(id)
 
- def set(self, id: _ID, data: _ANCHOR) -> None:
+ def set(self, data: _ANCHOR) -> None:
  """Save anchor to memory."""
- self.__mem__[id] = data
+ self.__mem__[data.jid] = data
 
- def remove(self, ids: _ID | Iterable[_ID]) -> None:
+ def remove(self, ids: JID[_ID, _ANCHOR] | Iterable[JID[_ID, _ANCHOR]]) -> None:
  """Remove anchor/s from memory."""
  if not isinstance(ids, Iterable):
  ids = [ids]
@@ -253,12 +266,12 @@ class ExecutionContext(Generic[_NODE_ANCHOR], ABC):
  @abstractmethod
  def init_anchor(
  self,
- anchor_jid: JID | None,
+ anchor_jid: str | None,
  default: _NODE_ANCHOR,
  ) -> _NODE_ANCHOR:
  """Load initial anchors."""
 
- def set_entry_node(self, entry_node: JID | None) -> None:
+ def set_entry_node(self, entry_node: str | None) -> None:
  """Override entry."""
  self.entry_node = self.init_anchor(entry_node, self.root)
 

jac/jaclang/runtimelib/memory.py

@@ -7,14 +7,20 @@
 from shelve import Shelf, open
 from typing import Callable, Generator, Iterable, TypeVar
 
-from .implementation import Anchor, JID, NodeAnchor, Root
+from .implementation import (
+ Anchor,
+ EdgeAnchor,
+ JID,
+ NodeAnchor,
+ Root,
+ WalkerAnchor,
+ _ANCHOR,
+)
 from .interface import Memory
 
-ID = TypeVar("ID")
-
 
 @dataclass
-class ShelfStorage(Memory[JID[Anchor], Anchor]):
+class ShelfStorage(Memory[JID, Anchor]):
  """Shelf Handler."""
 
  __shelf__: Shelf[Anchor] | None = None
@@ -69,9 +75,9 @@ def close(self) -> None:
 
  def find(
  self,
- ids: JID[Anchor] | Iterable[JID[Anchor]],
- filter: Callable[[Anchor], Anchor] | None = None,
- ) -> Generator[Anchor, None, None]:
+ ids: JID[_ANCHOR] | Iterable[JID[_ANCHOR]],
+ filter: Callable[[_ANCHOR], _ANCHOR] | None = None,
+ ) -> Generator[_ANCHOR, None, None]:
  """Find anchors from datasource by ids with filter."""
  if not isinstance(ids, Iterable):
  ids = [ids]
@@ -91,7 +97,7 @@ def find(
  else:
  yield from super().find(ids, filter)
 
- def find_by_id(self, id: JID[Anchor]) -> Anchor | None:
+ def find_by_id(self, id: JID[_ANCHOR]) -> _ANCHOR | None:
  """Find one by id."""
  data = super().find_by_id(id)