update ingestion to use ChannelConfig instead of Channel

Author: alexluck-sift

python/lib/sift_client/resources/rules.py

@@ -156,7 +156,9 @@ async def update(
         if isinstance(update, dict):
             update = RuleUpdate.model_validate(update)
 
-        updated_rule = await self._low_level_client.update_rule(rule=rule_obj, update=update, version_notes=version_notes)
+        updated_rule = await self._low_level_client.update_rule(
+            rule=rule_obj, update=update, version_notes=version_notes
+        )
         return self._apply_client_to_instance(updated_rule)
 
     async def archive(self, rule: str | Rule) -> Rule:

python/lib/sift_client/sift_types/__init__.py

@@ -9,7 +9,7 @@
     ChannelDataType,
     ChannelReference,
 )
-from sift_client.sift_types.ingestion import IngestionConfig
+from sift_client.sift_types.ingestion import ChannelConfig, Flow, IngestionConfig
 from sift_client.sift_types.rule import (
     Rule,
     RuleAction,
@@ -27,8 +27,10 @@
     "CalculatedChannelUpdate",
     "Channel",
     "ChannelBitFieldElement",
+    "ChannelConfig",
     "ChannelDataType",
     "ChannelReference",
+    "Flow",
     "IngestionConfig",
     "Rule",
     "RuleAction",

python/lib/sift_client/sift_types/_base.py

@@ -18,7 +18,7 @@ class BaseType(BaseModel, Generic[ProtoT, SelfT], ABC):
     model_config = ConfigDict(frozen=True)
 
     id_: str | None = None
-    proto: Any | None = Field(default=None, exclude=True) # For user reference only
+    proto: Any | None = Field(default=None, exclude=True)  # For user reference only
     _client: SiftClient | None = None
 
     @property

python/lib/sift_client/sift_types/channel.py

@@ -24,7 +24,6 @@
     Uint32Values,
     Uint64Values,
 )
-from sift.ingestion_configs.v2.ingestion_configs_pb2 import ChannelConfig
 
 from sift_client.sift_types._base import BaseType
 
@@ -235,9 +234,7 @@ def _enum_types_from_proto_list(enum_types: list[ChannelEnumTypePb]) -> dict[str
         return {enum.name: enum.key for enum in enum_types}
 
     @classmethod
-    def _from_proto(
-        cls, proto: ChannelProto, sift_client: SiftClient | None = None
-    ) -> Channel:
+    def _from_proto(cls, proto: ChannelProto, sift_client: SiftClient | None = None) -> Channel:
         return cls(
             proto=proto,
             id_=proto.channel_id,
@@ -257,19 +254,6 @@ def _from_proto(
             _client=sift_client,
         )
 
-
-    def _to_config_proto(self) -> ChannelConfig:
-        return ChannelConfig(
-            name=self.name,
-            data_type=self.data_type.value,
-            description=self.description,  # type: ignore
-            unit=self.unit,  # type: ignore
-            bit_field_elements=[el._to_proto() for el in self.bit_field_elements]
-            if self.bit_field_elements
-            else None,
-            enum_types=self._enum_types_to_proto_list(self.enum_types),
-        )
-
     def data(
         self,
         *,

python/lib/sift_client/sift_types/ingestion.py

@@ -4,8 +4,11 @@
 from typing import TYPE_CHECKING, Any
 
 from google.protobuf.empty_pb2 import Empty
-from pydantic import ConfigDict
+from pydantic import ConfigDict, model_validator
 from sift.ingest.v1.ingest_pb2 import IngestWithConfigDataChannelValue
+from sift.ingestion_configs.v2.ingestion_configs_pb2 import (
+    ChannelConfig as ChannelConfigProto,
+)
 from sift.ingestion_configs.v2.ingestion_configs_pb2 import (
     FlowConfig,
 )
@@ -22,12 +25,13 @@
 )
 
 from sift_client.sift_types._base import BaseType
-from sift_client.sift_types.channel import Channel, ChannelDataType
+from sift_client.sift_types.channel import ChannelBitFieldElement, ChannelDataType
 
 if TYPE_CHECKING:
     from datetime import datetime
 
     from sift_client.client import SiftClient
+    from sift_client.sift_types.channel import Channel
 
 
 class IngestionConfig(BaseType[IngestionConfigProto, "IngestionConfig"]):
@@ -49,6 +53,98 @@ def _from_proto(
         )
 
 
+class ChannelConfig(BaseType[ChannelConfigProto, "ChannelConfig"]):
+    """Channel configuration model for ingestion purposes.
+
+    This model contains only the fields needed for ingestion configuration,
+    without the full metadata from the Channels API.
+    """
+
+    model_config = ConfigDict(frozen=False)
+    name: str
+    data_type: ChannelDataType
+    description: str | None = None
+    unit: str | None = None
+    bit_field_elements: list[ChannelBitFieldElement] | None = None
+    enum_types: dict[str, int] | None = None
+
+    @model_validator(mode="after")
+    def _validate_enum_types(self):
+        """Validate that enum_types is provided when data_type is ENUM."""
+        if self.data_type == ChannelDataType.ENUM and not self.enum_types:
+            raise ValueError(
+                f"Channel '{self.name}' has data_type ENUM but enum_types is not provided"
+            )
+        return self
+
+    @classmethod
+    def _from_proto(
+        cls, proto: ChannelConfigProto, sift_client: SiftClient | None = None
+    ) -> ChannelConfig:
+        """Create ChannelConfig from ChannelConfigProto."""
+        return cls(
+            proto=proto,
+            name=proto.name,
+            data_type=ChannelDataType(proto.data_type),
+            description=proto.description if proto.description else None,
+            unit=proto.unit if proto.unit else None,
+            bit_field_elements=[
+                ChannelBitFieldElement._from_proto(el) for el in proto.bit_field_elements
+            ]
+            if proto.bit_field_elements
+            else None,
+            enum_types={enum.name: enum.key for enum in proto.enum_types}
+            if proto.enum_types
+            else None,
+            _client=sift_client,
+        )
+
+    @classmethod
+    def from_channel(cls, channel: Channel) -> ChannelConfig:
+        """Create ChannelConfig from a Channel.
+
+        Args:
+            channel: The Channel to convert.
+
+        Returns:
+            A ChannelConfig with the channel's configuration data.
+        """
+        return cls(
+            name=channel.name,
+            data_type=channel.data_type,
+            description=channel.description,
+            unit=channel.unit,
+            bit_field_elements=channel.bit_field_elements if channel.bit_field_elements else None,
+            enum_types=channel.enum_types,
+        )
+
+    def _to_config_proto(self) -> ChannelConfigProto:
+        """Convert to ChannelConfigProto for ingestion."""
+        from sift.common.type.v1.channel_bit_field_element_pb2 import (
+            ChannelBitFieldElement as ChannelBitFieldElementPb,
+        )
+        from sift.common.type.v1.channel_enum_type_pb2 import ChannelEnumType as ChannelEnumTypePb
+
+        return ChannelConfigProto(
+            name=self.name,
+            data_type=self.data_type.value,
+            description=self.description or "",
+            unit=self.unit or "",
+            bit_field_elements=[
+                ChannelBitFieldElementPb(
+                    name=bfe.name,
+                    index=bfe.index,
+                    bit_count=bfe.bit_count,
+                )
+                for bfe in self.bit_field_elements or []
+            ],
+            enum_types=[
+                ChannelEnumTypePb(name=name, key=key)
+                for name, key in (self.enum_types or {}).items()
+            ],
+        )
+
+
 class Flow(BaseType[FlowConfig, "Flow"]):
     """Model representing a data flow for ingestion.
 
@@ -57,7 +153,7 @@ class Flow(BaseType[FlowConfig, "Flow"]):
 
     model_config = ConfigDict(frozen=False)
     name: str
-    channels: list[Channel]
+    channels: list[ChannelConfig]
     ingestion_config_id: str | None = None
     run_id: str | None = None
 
@@ -66,7 +162,7 @@ def _from_proto(cls, proto: FlowConfig, sift_client: SiftClient | None = None) -
         return cls(
             proto=proto,
             name=proto.name,
-            channels=[Channel._from_proto(channel) for channel in proto.channels],
+            channels=[ChannelConfig._from_proto(channel) for channel in proto.channels],
             _client=sift_client,
         )
 
@@ -82,11 +178,11 @@ def _to_rust_config(self) -> FlowConfigPy:
             channels=[_channel_to_rust_config(channel) for channel in self.channels],
         )
 
-    def add_channel(self, channel: Channel):
-        """Add a Channel to this Flow.
+    def add_channel(self, channel: ChannelConfig):
+        """Add a ChannelConfig to this Flow.
 
         Args:
-            channel: The Channel to add.
+            channel: The ChannelConfig to add.
 
         Raises:
             ValueError: If the flow has already been created with an ingestion config.
@@ -115,7 +211,7 @@ def ingest(self, *, timestamp: datetime, channel_values: dict[str, Any]):
 
 
 # Converter functions.
-def _channel_to_rust_config(channel: Channel) -> ChannelConfigPy:
+def _channel_to_rust_config(channel: ChannelConfig) -> ChannelConfigPy:
     return ChannelConfigPy(
         name=channel.name,
         data_type=_to_rust_type(channel.data_type),
@@ -135,7 +231,7 @@ def _channel_to_rust_config(channel: Channel) -> ChannelConfigPy:
 
 
 def _rust_channel_value_from_bitfield(
-    channel: Channel, value: Any
+    channel: ChannelConfig, value: Any
 ) -> IngestWithConfigDataChannelValuePy:
     """Helper function to convert a bitfield value to a ChannelValuePy object.
 
@@ -171,10 +267,10 @@ def _rust_channel_value_from_bitfield(
     return IngestWithConfigDataChannelValuePy.bitfield(byte_array)
 
 
-def _to_rust_value(channel: Channel, value: Any) -> IngestWithConfigDataChannelValuePy:
+def _to_rust_value(channel: ChannelConfig, value: Any) -> IngestWithConfigDataChannelValuePy:
     if value is None:
         return IngestWithConfigDataChannelValuePy.empty()
-    if channel.data_type == ChannelDataType.ENUM:
+    if channel.data_type == ChannelDataType.ENUM and channel.enum_types is not None:
         enum_name = value
         enum_val = channel.enum_types.get(enum_name)
         if enum_val is None: