0.4.0 release

.gitignore

@@ -8,3 +8,7 @@ venv/
 .venv/
 
 dist/
+
+*.lock
+test.py
+pyrightconfig.json

TODO.md

@@ -6,5 +6,21 @@
 - [x] Proper logging
 - [x] Proper limits handling
 - [x] Try reconnecting on initial connect failure
+- [ ] JetStream
+    1. - [x] Pub, Pull/Push subscription, Ack, Nak, etc.
+    2. - [x] Flow control for push consumers
+    3. - [x] DIRECT get message
+    4. - [x] Ordered push consumers
+    5. - [x] KV storage
+    6. - [ ] Object storage
+- [ ] Tests with devcontainers
+    1. - [x] NATS Core
+    2. - [x] Protocol parser
+    3. - [x] Headers parser
+    4. - [ ] Different connections
+    5. - [ ] NATS JetStream (pull/push sub, kv/object storage)
 - [ ] Auth (JWT, User/Pass, NKey)
-- [ ] JetStream (Pub, Pull/Push subscription, Ack, etc.)
+- [ ] Micro
+- [ ] Cluster support (cluster endpoints discovery, update server info on `INFO` operation message, etc.)
+- [x] Rework serialization of JS api
+- [ ] Try to use mixins for subscriptions

natsio/__init__.py

@@ -1 +1 @@
-__version__ = "0.2.1"
+__version__ = "0.3.0"

natsio/_internal/serialization/converters.py

@@ -0,0 +1,28 @@
+from datetime import datetime, timedelta
+from typing import Final
+
+from natsio.utils.time import fromisoformat
+
+from .types import Converter
+
+NANOSECOND_POWER: Final[int] = 10**9
+
+
+class TimedeltaNanosecondsConverter(Converter[timedelta, int]):
+    def to_wire(self, value: timedelta) -> int:
+        return int(value.total_seconds() * NANOSECOND_POWER)
+
+    def from_wire(self, value: int) -> timedelta:
+        return timedelta(microseconds=value / 1000)
+
+
+class DatetimeIsoConverter(Converter[datetime, str]):
+    def to_wire(self, value: datetime) -> str:
+        return value.isoformat()
+
+    def from_wire(self, value: str) -> datetime:
+        return fromisoformat(value)
+
+
+TIMEDELTA_NANO = TimedeltaNanosecondsConverter()
+DATETIME_ISO = DatetimeIsoConverter()

natsio/_internal/serialization/serializator.py

@@ -0,0 +1,143 @@
+from collections.abc import Mapping as MappingABC
+from dataclasses import fields, is_dataclass
+from enum import Enum
+from types import UnionType
+from typing import (
+    Annotated,
+    Any,
+    Mapping,
+    Union,
+    cast,
+    get_args,
+    get_origin,
+    get_type_hints,
+)
+
+from .types import DT, Converter, DataclassInstance
+
+
+def get_converter_from_annotation(
+    field_type: type,
+) -> Converter[Any, Any] | None:
+    if get_origin(field_type) is Annotated:
+        args = get_args(field_type)
+        for arg in args[1:]:
+            if isinstance(arg, Converter):
+                return arg
+    return None
+
+
+def get_base_type(field_type: type) -> type:
+    origin = get_origin(field_type)
+    args = get_args(field_type)
+    if origin is Annotated:
+        return cast(type, args[0])
+
+    if origin is not None:
+        if origin is UnionType or origin is Union:
+            for arg in args:
+                if arg is not type(None):
+                    return get_base_type(arg)
+        return cast(type, origin)
+    return field_type
+
+
+def handle_enum(value: Any, enum_type: type) -> Any:
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, str) and issubclass(enum_type, Enum):
+        try:
+            return enum_type(value)
+        except ValueError:
+            raise ValueError(f"Invalid value '{value}' for enum {enum_type.__name__}")
+    return value
+
+
+def serialize_value(value: Any, field_type: type) -> Any:
+    if value is None:
+        return None
+
+    converter = get_converter_from_annotation(field_type)
+    if converter is not None:
+        return converter.to_wire(value)
+
+    base_type = get_base_type(field_type)
+    origin = get_origin(field_type)
+    args = get_args(field_type)
+
+    if isinstance(value, Enum):
+        return handle_enum(value, base_type)
+
+    if origin is list and args:
+        item_type = args[0] if args else type(Any)
+        return [serialize_value(val, item_type) for val in value]
+
+    if origin is MappingABC and len(args) == 2:
+        item_type = args[1] if args else type(Any)
+        return {k: serialize_value(v, item_type) for k, v in value.items()}
+
+    if is_dataclass(value.__class__):
+        return serialize_dataclass(value)
+
+    return value
+
+
+def deserialize_value(value: Any, field_type: type) -> Any:
+    if value is None:
+        return None
+
+    converter = get_converter_from_annotation(field_type)
+    if converter is not None:
+        return converter.from_wire(value)
+
+    base_type = get_base_type(field_type)
+    origin = get_origin(field_type)
+    args = get_args(field_type)
+
+    if is_dataclass(base_type):
+        return deserialize_dataclass(value, base_type)
+
+    if origin is list and args:
+        item_type = args[0] if args else type(Any)
+        return [deserialize_value(val, item_type) for val in value]
+
+    if origin is MappingABC and len(args) == 2:
+        item_type = args[1] if args else type(Any)
+        return {k: deserialize_value(v, item_type) for k, v in value.items()}
+
+    if issubclass(base_type, Enum):
+        return handle_enum(value, base_type)
+
+    if isinstance(value, (str, int, float, bool)) and not isinstance(value, base_type):
+        try:
+            return base_type(value)  # type: ignore[call-arg]
+        except (ValueError, TypeError):
+            pass
+
+    return value
+
+
+def serialize_dataclass(obj: DataclassInstance) -> Mapping[str, Any]:
+    result = {}
+    type_hints = get_type_hints(obj.__class__, include_extras=True)
+
+    for field in fields(obj):
+        value = getattr(obj, field.name)
+        if value is not None or field.default is None or field.default_factory is None:
+            result[field.name] = serialize_value(value, type_hints[field.name])
+
+    return result
+
+
+def deserialize_dataclass(data: Mapping[str, Any], cls: type[DT]) -> DT:
+    type_hints = get_type_hints(cls, include_extras=True)
+    kwargs = {}
+
+    for field in fields(cls):
+        if field.name in data:
+            kwargs[field.name] = deserialize_value(
+                data[field.name],
+                type_hints[field.name],
+            )
+
+    return cls(**kwargs)

natsio/_internal/serialization/types.py

@@ -0,0 +1,18 @@
+from dataclasses import Field
+from typing import Any, ClassVar, Protocol, TypeVar, runtime_checkable
+
+
+class DataclassInstance(Protocol):
+    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]
+
+
+DT = TypeVar("DT", bound=DataclassInstance)
+
+T = TypeVar("T")
+V = TypeVar("V")
+
+
+@runtime_checkable
+class Converter(Protocol[T, V]):
+    def to_wire(self, value: T) -> V: ...
+    def from_wire(self, value: V) -> T: ...

natsio/abc/connection.py

@@ -1,17 +1,16 @@
 import asyncio
 from ssl import SSLContext
-from typing import TYPE_CHECKING, Optional, Protocol
+from typing import Protocol, Self
 
+from natsio.config import ServerInfo
 from natsio.connection.status import ConnectionStatus
 from natsio.protocol.operations.connect import Connect
 
 from .client import ErrorCallback
 from .dispatcher import DispatcherProto
+from .json import JSONSerializerProto
 from .protocol import ClientMessageProto
 
-if TYPE_CHECKING:
-    from natsio.client.config import ServerInfo
-
 
 class StreamProto(Protocol):
     @property
@@ -67,7 +66,7 @@ def outstanding_pings(self) -> int:
         raise NotImplementedError
 
     @property
-    def server_info(self) -> "ServerInfo":
+    def server_info(self) -> ServerInfo:
         raise NotImplementedError
 
     @classmethod
@@ -85,15 +84,14 @@ async def connect(
         force_flush_timeout: int,
         error_callback: ErrorCallback,
         timeout: float,
-        ssl: Optional[SSLContext],
-        ssl_hostname: Optional[str],
-        handshake_first: Optional[bool],
-    ) -> "ConnectionProto":
+        json_serializer: JSONSerializerProto,
+        ssl: SSLContext | None,
+        ssl_hostname: str | None,
+        handshake_first: bool | None,
+    ) -> Self:
         raise NotImplementedError
 
-    async def send_command(
-        self, cmd: ClientMessageProto, force_flush: bool = False
-    ) -> None:
+    async def send_command(self, cmd: ClientMessageProto, force_flush: bool = False) -> None:
         raise NotImplementedError
 
     async def flush(self) -> None:

natsio/abc/dispatcher.py

@@ -1,17 +1,17 @@
 import asyncio
 from typing import Protocol, Sequence
 
+from natsio.abc.subscription import SubscriptionProto
 from natsio.messages.core import CoreMsg
 from natsio.protocol.operations.hmsg import HMsg
 from natsio.protocol.operations.msg import Msg
-from natsio.subscriptions.core import Subscription
 
 
 class DispatcherProto(Protocol):
-    def add_subscription(self, sub: Subscription) -> None:
+    def add_subscription(self, sub: SubscriptionProto) -> None:
         raise NotImplementedError
 
-    def all_subscriptions(self) -> Sequence[Subscription]:
+    def all_subscriptions(self) -> Sequence[SubscriptionProto]:
         raise NotImplementedError
 
     def remove_subscription(self, sid: str) -> None:

natsio/abc/json.py

@@ -0,0 +1,9 @@
+from typing import Any, Protocol
+
+
+class JSONSerializerProto(Protocol):
+    def load(self, obj: str | bytes) -> Any:
+        raise NotImplementedError
+
+    def dump(self, obj: Any) -> bytes:
+        raise NotImplementedError

natsio/abc/subscription.py

@@ -0,0 +1,48 @@
+from enum import Enum
+from typing import TypeVar, Protocol
+
+from natsio.messages.core import CoreMsg
+from natsio.messages.jetstream import JetStreamMsg
+
+
+MsgType = TypeVar("MsgType", CoreMsg, JetStreamMsg, contravariant=True)
+
+
+class Callback(Protocol[MsgType]):
+    async def __call__(self, msg: MsgType) -> None:
+        raise NotImplementedError
+
+
+CoreCallback = Callback[CoreMsg]
+JetStreamCallback = Callback[JetStreamMsg]
+
+
+class SubscriptionStatus(Enum):
+    INITIALISING = "INITIALISING"
+    OPERATING = "OPERATING"
+    DRAINING = "DRAINING"
+    CLOSED = "CLOSED"
+
+
+class SubscriptionProto(Protocol):
+    sid: str
+    subject: str
+    queue: str | None
+
+    @property
+    def is_ready_to_close(self) -> bool:
+        raise NotImplementedError
+
+    async def add_msg(self, msg: CoreMsg) -> None:
+        raise NotImplementedError
+
+    async def unsubscribe(self, max_msgs: int = 0) -> None:
+        raise NotImplementedError
+
+
+__all__ = (
+    "CoreCallback",
+    "JetStreamCallback",
+    "SubscriptionStatus",
+    "SubscriptionProto",
+)

natsio/client/__init__.py

@@ -1,3 +1,7 @@
 from .core import NATSCore
-from .config import ClientConfig
 from .status import ClientStatus
+
+__all__ = (
+    "NATSCore",
+    "ClientStatus",
+)