feat: Encoding (#143)

* feat: Add encoding types

* feat: Add encoding logic to publish_task

* chore: Bump Python SDK version

* feat: Modify models to be explicit about task input being raw

* fix: Refactor encoding base types to be classes instead of just functions

* feat: Add Passthrough encoders

* feat: Add Pydantic encoders

* chore: Update encoder exports

* feat: Improve worker handler structure, typing, and add encoding

* test: Add encoder tests

* test: Update tests to account for new encoding ergonomics

* fix: Fix poor type handling in Avro and Pydantic task update schemas

* chore: Bump package version

* chore: Sync updated Avro schemas

* test: Update e2e tests to account for new encoders

* fix: Fix bug causing task update data to be possibly mixed up and corrupted by adding a unique constant identifier that gets validated on ingestion

* chore: Bump cargo version

* chore: Bump all versions to 0.4.0

* docs: Update docs to account for new changes

* chore: Fix versions for alpha release

* feat: Add custom encoding error

* feat: Add JSON dictionary encoders

* feat: Add improved error handling to Pydantic encoders

* feat: Add string encoders

* feat: Add polymorphic encoder factory so we can adapt to the most common types

* docs: Update task documentation to use polymorphic encoder

* test: Remove explicit encoders from e2e tests

* test: Add tests for all new encoders

* test: Create a new task for user-friendly error

* feat: Add new type-hinting based encoding logic to publisher and worker

* feat: Export all default encoders in Taco

* docs: Remove most explicit encoding from setup page

* Update .gitignore

Author: tubarao312

.gitignore

@@ -15,3 +15,5 @@
 .benchmarks/
 __pycache__/
 /client_sdks/python/tacoq.egg-info
+.vscode/settings.json
+client_sdks/python/.vscode/settings.json

Cargo.lock

@@ -1,6 +1,6 @@
 [project]
 name = "tacoq"
-version = "0.3.0b"
+version = "0.4.0-alpha.1"
 description = "Python SDK to interact with the TacoQ task queue."
 readme = "README.md"
 authors = [
@@ -22,7 +22,6 @@ dependencies = [
     "fastavro>=1.10.0",
     "opentelemetry-api>=1.30.0",
     "pydantic>=2.10.5",
-    "pytest-timeout>=2.3.1",
     "tenacity>=9.0.0",
     "uuid>=1.30",
     "watchfiles>=1.0.3",
@@ -61,6 +60,7 @@ dev = [
     "pytest>=8.3.5",
     "pytest-asyncio>=0.25.3",
     "pytest-xdist>=3.6.1",
+    "pytest-timeout>=2.3.1",
     "opentelemetry-sdk>=1.30.0",
     "opentelemetry-exporter-otlp>=1.30.0",
 ]

client_sdks/python/pyproject.toml

@@ -1,8 +1,23 @@
+from tacoq.core.encoding import (
+    Data,
+    Decoder,
+    Encoder,
+    PassthroughDecoder,
+    PassthroughEncoder,
+    PydanticDecoder,
+    PydanticEncoder,
+    StringDecoder,
+    StringEncoder,
+    JsonDictDecoder,
+    JsonDictEncoder,
+    create_encoder,
+    create_decoder,
+)
 from tacoq.core.infra.broker import BrokerConfig
-from tacoq.relay import RelayConfig, RelayClient, RelayStates
-from tacoq.core.models import Task, TaskInput, TaskOutput
-from tacoq.core.telemetry import TracerManager, LoggerManager
+from tacoq.core.models import Task, TaskRawInput, TaskRawOutput
+from tacoq.core.telemetry import LoggerManager, TracerManager
 from tacoq.publisher import PublisherClient
+from tacoq.relay import RelayClient, RelayConfig, RelayStates
 from tacoq.worker import WorkerApplication, WorkerApplicationConfig
 
 __all__ = [
@@ -14,8 +29,21 @@
     "WorkerApplication",
     "WorkerApplicationConfig",
     "Task",
-    "TaskInput",
-    "TaskOutput",
+    "TaskRawInput",
+    "TaskRawOutput",
     "TracerManager",
     "LoggerManager",
+    "PydanticEncoder",
+    "PydanticDecoder",
+    "PassthroughEncoder",
+    "PassthroughDecoder",
+    "Encoder",
+    "Decoder",
+    "Data",
+    "StringEncoder",
+    "StringDecoder",
+    "JsonDictEncoder",
+    "JsonDictDecoder",
+    "create_encoder",
+    "create_decoder",
 ]

client_sdks/python/tacoq/__init__.py

@@ -0,0 +1,32 @@
+from tacoq.core.encoding.models import (
+    Encoder,
+    Decoder,
+    EncodingError,
+    Data,
+)
+
+from tacoq.core.encoding.pydantic import PydanticDecoder, PydanticEncoder
+from tacoq.core.encoding.passthrough import (
+    PassthroughDecoder,
+    PassthroughEncoder,
+)
+from tacoq.core.encoding.json_dict import JsonDictDecoder, JsonDictEncoder
+from tacoq.core.encoding.string import StringDecoder, StringEncoder
+from tacoq.core.encoding.polymorphic import create_encoder, create_decoder
+
+__all__ = [
+    "EncodingError",
+    "Encoder",
+    "Decoder",
+    "Data",
+    "create_encoder",
+    "create_decoder",
+    "PydanticDecoder",
+    "PydanticEncoder",
+    "PassthroughDecoder",
+    "PassthroughEncoder",
+    "JsonDictDecoder",
+    "JsonDictEncoder",
+    "StringDecoder",
+    "StringEncoder",
+]

client_sdks/python/tacoq/core/encoding/__init__.py

@@ -0,0 +1,65 @@
+"""Encoders/Decoders that translate between Python dictionaries and JSON bytes."""
+
+import json
+from typing import Dict, Any
+from tacoq.core.encoding.models import Encoder, Decoder, EncodingError
+
+
+class JsonDictEncoder(Encoder[Dict[str, Any]]):
+    """Encodes Python dictionaries to JSON bytes.
+
+    ### Usage:
+    ```python
+    from tacoq.core.encoding import JsonDictEncoder
+
+    encoder = JsonDictEncoder()
+    encoded_data = encoder.encode({"name": "John", "age": 30})
+    ```
+    """
+
+    def encode(self, data: Dict[str, Any]) -> bytes:
+        """Encodes a dictionary to JSON bytes.
+
+        ### Arguments:
+        - data: The dictionary to encode
+
+        ### Returns:
+        The encoded JSON bytes
+
+        ### Raises:
+        - EncodingError: If there's an error encoding the message
+        """
+        try:
+            return json.dumps(data).encode("utf-8")
+        except Exception as e:
+            raise EncodingError(f"Error encoding message: {str(e)}")
+
+
+class JsonDictDecoder(Decoder[Dict[str, Any]]):
+    """Decodes JSON bytes to Python dictionaries.
+
+    ### Usage:
+    ```python
+    from tacoq.core.encoding import JsonDictDecoder
+
+    decoder = JsonDictDecoder()
+    decoded_data = decoder.decode(b'{"name": "John", "age": 30}')
+    ```
+    """
+
+    def decode(self, data: bytes) -> Dict[str, Any]:
+        """Decodes JSON bytes to a dictionary.
+
+        ### Arguments:
+        - data: The JSON bytes to decode
+
+        ### Returns:
+        The decoded dictionary
+
+        ### Raises:
+        - EncodingError: If there's an error decoding the message
+        """
+        try:
+            return json.loads(data.decode("utf-8"))
+        except Exception as e:
+            raise EncodingError(f"Error decoding message: {str(e)}")

client_sdks/python/tacoq/core/encoding/json_dict.py

@@ -0,0 +1,28 @@
+"""Interfaces for byte encoders/decoders."""
+
+from typing import Generic, TypeVar
+
+Data = TypeVar("Data")
+""" The type of the data to encode/decode. """
+
+
+class Encoder(Generic[Data]):
+    """An encoder for a specific type of data."""
+
+    def encode(self, data: Data) -> bytes:
+        """Encodes an object from the output of the task into bytes."""
+        ...
+
+
+class Decoder(Generic[Data]):
+    """A decoder for a specific type of data."""
+
+    def decode(self, data: bytes) -> Data:
+        """Decodes an object from bytes into the input of the task."""
+        ...
+
+
+class EncodingError(Exception):
+    """Raised when there's an error encoding or decoding a message."""
+
+    pass

client_sdks/python/tacoq/core/encoding/models.py

@@ -0,0 +1,56 @@
+"""Generic passthrough decoders that you can use if you need your encoding
+logic to be more complex than what can fit within the standard `Encoder`
+and `Decoder` interfaces.
+"""
+
+from tacoq.core.encoding.models import Encoder, Decoder
+
+
+class PassthroughEncoder(Encoder[bytes]):
+    """Passthrough encoder that returns the input bytes unchanged.
+
+    ### Usage:
+    ```python
+    from tacoq.core.encoding import PassthroughEncoder
+
+    encoder = PassthroughEncoder()
+    encoded_data = encoder.encode(b"Hello, world!")
+    # encoded_data will be b"Hello, world!"
+    ```
+    """
+
+    def encode(self, data: bytes) -> bytes:
+        """Passthrough encoder that returns the input bytes unchanged.
+
+        ### Arguments:
+        - data: The bytes to encode
+
+        ### Returns:
+        The same bytes unchanged
+        """
+        return data
+
+
+class PassthroughDecoder(Decoder[bytes]):
+    """Passthrough decoder that returns the input bytes unchanged.
+
+    ### Usage:
+    ```python
+    from tacoq.core.encoding import PassthroughDecoder
+
+    decoder = PassthroughDecoder()
+    decoded_data = decoder.decode(b"Hello, world!")
+    # decoded_data will be b"Hello, world!"
+    ```
+    """
+
+    def decode(self, data: bytes) -> bytes:
+        """Passthrough decoder that returns the input bytes unchanged.
+
+        ### Arguments:
+        - data: The bytes to decode
+
+        ### Returns:
+        The same bytes unchanged
+        """
+        return data

client_sdks/python/tacoq/core/encoding/passthrough.py

@@ -0,0 +1,69 @@
+"""Factory functions for creating encoders and decoders based on type."""
+
+from inspect import isclass
+from typing import Any, Type, TypeVar, Union, cast, get_origin
+
+from pydantic import BaseModel
+from tacoq.core.encoding.json_dict import JsonDictDecoder, JsonDictEncoder
+from tacoq.core.encoding.models import Decoder, Encoder, EncodingError
+from tacoq.core.encoding.passthrough import PassthroughDecoder, PassthroughEncoder
+from tacoq.core.encoding.pydantic import PydanticDecoder, PydanticEncoder
+from tacoq.core.encoding.string import StringDecoder, StringEncoder
+
+SUPPORTED_DATA_TYPES = (bytes, str, dict, list, BaseModel)
+
+T = TypeVar("T", bound=Union[bytes, str, dict[str, Any], list[Any], BaseModel])
+
+
+def create_encoder(data_type: Type[T]) -> Encoder[T]:
+    """Creates an appropriate encoder based on the input type.
+
+    ### Arguments:
+    - data_type: The type to create an encoder for
+
+    ### Returns:
+    An encoder instance appropriate for the given type
+
+    ### Raises:
+    - EncodingError: If no suitable encoder is found for the type
+    """
+
+    if data_type is bytes:
+        return cast(Encoder[T], PassthroughEncoder())
+    elif data_type is str:
+        return cast(Encoder[T], StringEncoder())
+    elif data_type is dict or get_origin(data_type) is dict:
+        return cast(Encoder[T], JsonDictEncoder())
+    elif isclass(data_type) and issubclass(data_type, BaseModel):
+        return cast(Encoder[T], PydanticEncoder())
+    else:
+        raise EncodingError(
+            f"No encoder found for type {data_type}. Available encoder types: {SUPPORTED_DATA_TYPES}. If yours isn't one of these, perhaps you should implement your own encoder?"
+        )
+
+
+def create_decoder(data_type: Type[T]) -> Decoder[T]:
+    """Creates an appropriate decoder based on the input type.
+
+    ### Arguments:
+    - data_type: The type to create a decoder for
+
+    ### Returns:
+    A decoder instance appropriate for the given type
+
+    ### Raises:
+    - EncodingError: If no suitable decoder is found for the type
+    """
+
+    if data_type is bytes:
+        return cast(Decoder[T], PassthroughDecoder())
+    elif data_type is str:
+        return cast(Decoder[T], StringDecoder())
+    elif data_type is dict or get_origin(data_type) is dict:
+        return cast(Decoder[T], JsonDictDecoder())
+    elif issubclass(data_type, BaseModel):
+        return cast(Decoder[T], PydanticDecoder(model=data_type))
+    else:
+        raise EncodingError(
+            f"No decoder found for type {data_type}. Available decoder types: {SUPPORTED_DATA_TYPES}. If yours isn't one of these, perhaps you should implement your own decoder?"
+        )