Orc 50 - Simplified the way the processor address is supplied

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [unreleased]
 
+### Changed
+
+- How the processor address is provided, with a robust regex handling of the address
+
+## [v0.8.0] - 23-09-2025
+
 ### Added
 
 - Integrated support for more robust metadata handling

examples/simpletrigger/processor.py

@@ -10,22 +10,23 @@
 
 proc = Processor("ml")
 
-trip_id = MetadataField(name="trip_id", description="The unique ID of the trip")
+# E.g. you have you have a fleet of busses, where every bus has a particular ID and runs a particular route
+route_id = MetadataField(name="route_id", description="The unique ID of the route")
 bus_id = MetadataField(name="bus_id", description="The unique ID of the bus")
 
 Every30Second = WindowType(
     name="Every30Second",
     version="1.0.0",
     description="Triggers every 30 seconds",
-    metadataFields=[trip_id, bus_id],
+    metadataFields=[route_id, bus_id],
 )
 
 
 @proc.algorithm("MyAlgo", "1.0.0", Every30Second)
 def my_algorithm(params: ExecutionParams) -> StructResult:
-    trip_id = params.window.metadata.get("trip_id", None)
+    route_id = params.window.metadata.get("route_id", None)
     bus_id = params.window.metadata.get("bus_id", None)
-    print(trip_id, bus_id)
+    print(route_id, bus_id)
 
     time.sleep(5)
     return StructResult({"result": 42})

examples/simpletrigger/window.py

@@ -2,7 +2,7 @@
 import datetime as dt
 
 import schedule
-from processor import bus_id, trip_id
+from processor import bus_id, route_id
 
 from orca_python import Window, EmitWindow
 
@@ -15,7 +15,7 @@ def emitWindow() -> None:
         name="Every30Second",
         version="1.0.0",
         origin="Example",
-        metadata={bus_id.name: 1, trip_id.name: 2},
+        metadata={bus_id.name: 1, route_id.name: 2},
     )
     EmitWindow(window)
 

orca_python/envs.py

@@ -1,30 +1,62 @@
 import os
-from typing import Tuple
+import re
+from typing import Tuple, Optional
 
-from orca_python.exceptions import MissingDependency
+from orca_python.exceptions import BadEnvVar, MissingEnvVar
 
 
-def getenvs() -> Tuple[bool, str, str, str]:
-    orcaserver = os.getenv("ORCA_CORE", "")
-    if orcaserver == "":
-        raise MissingDependency("ORCA_CORE is required")
-    orcaserver = orcaserver.lstrip("grpc://")
+def _parse_connection_string(connection_string: str) -> Optional[Tuple[str, int]]:
+    """
+    Parse a connection string of the form 'address:port'.
+    """
+    # check for leading/trailing whitespace
+    if connection_string != connection_string.strip():
+        return None
 
-    port = os.getenv("PROCESSOR_PORT", "")
-    if port == "":
-        raise MissingDependency("PROCESSOR_PORT required")
+    # use regex to find the port at the end
+    port_pattern = r":(\d+)$"
+    match = re.search(port_pattern, connection_string)
 
-    host = os.getenv("PROCESSOR_ADDRESS", "")
-    if host == "":
-        raise MissingDependency("PROCESSOR_ADDRESS is required")
+    if not match:
+        return None
+
+    # extract port
+    port = int(match.group(1))
+
+    # get address by removing the port part
+    address = connection_string[: match.start()]
+
+    # valdate that address is not empty
+    if not address:
+        return None
+
+    return (address, port)
+
+
+def getenvs() -> Tuple[bool, str, str, int]:
+    orca_core = os.getenv("ORCA_CORE", "")
+    if orca_core == "":
+        raise MissingEnvVar("ORCA_CORE is required")
+    orca_core = orca_core.lstrip("grpc://")
+
+    proc_address = os.getenv("PROCESSOR_ADDRESS", "")
+    if proc_address == "":
+        raise MissingEnvVar("PROCESSOR_ADDRESS is required")
+
+    res = _parse_connection_string(proc_address)
+    if res is None:
+        raise BadEnvVar(
+            "PROCESSOR_ADDRESS is not a valid address of the form <ip>:<port>"
+        )
+    processor_address, processor_port = res
 
     env = os.getenv("ENV", "")
     if env == "production":
         is_production = True
     else:
         is_production = False
 
-    return is_production, orcaserver, port, host
+    return is_production, orca_core, processor_address, processor_port
 
 
-is_production, ORCACORE, PORT, HOST = getenvs()
+is_production, ORCA_CORE, PROCESSOR_HOST, PROCESSOR_PORT = getenvs()

orca_python/exceptions.py

@@ -22,5 +22,9 @@ class InvalidDependency(BaseOrcaException):
     """Raised when a dependency is invalid"""
 
 
-class MissingDependency(BaseOrcaException):
-    """Raised when a dependency is missing"""
+class MissingEnvVar(BaseOrcaException):
+    """Raised when an environment variable is missing"""
+
+
+class BadEnvVar(BaseOrcaException):
+    """Raised when an environment variable is poorly defined"""

orca_python/main.py

@@ -239,14 +239,14 @@ def EmitWindow(window: Window) -> None:
     if envs.is_production:
         # secure channel with TLS
         with grpc.secure_channel(
-            envs.ORCACORE, grpc.ssl_channel_credentials()
+            envs.ORCA_CORE, grpc.ssl_channel_credentials()
         ) as channel:
             stub = service_pb2_grpc.OrcaCoreStub(channel)
             response = stub.EmitWindow(window_pb)
             LOGGER.info(f"Window emitted: {response}")
     else:
         # insecure channel for local development
-        with grpc.insecure_channel(envs.ORCACORE) as channel:
+        with grpc.insecure_channel(envs.ORCA_CORE) as channel:
             stub = service_pb2_grpc.OrcaCoreStub(channel)
             response = stub.EmitWindow(window_pb)
             LOGGER.info(f"Window emitted: {response}")
@@ -391,12 +391,8 @@ class Processor(OrcaProcessorServicer):  # type: ignore
     def __init__(self, name: str, max_workers: int = 10):
         super().__init__()
         self._name = name
-        self._processorConnStr = (
-            f"[::]:{envs.PORT}"  # attach the processor to all network interfaces.
-        )
-        self._orcaProcessorConnStr = (
-            envs.HOST
-        )  # tell orca-core to reference this processor by this address.
+        self._processorConnStr = f"[::]:{envs.PROCESSOR_PORT}"  # attach the processor to all network interfaces when launching the gRPC service.
+        self._orcaProcessorConnStr = f"{envs.PROCESSOR_HOST}:{envs.PROCESSOR_PORT}"  # tell orca-core to reference this processor by this address.
         self._runtime = sys.version
         self._max_workers = max_workers
         self._algorithmsSingleton: Algorithms = Algorithms()
@@ -687,14 +683,14 @@ def Register(self) -> None:
         if envs.is_production:
             # secure channel with TLS
             with grpc.secure_channel(
-                envs.ORCACORE, grpc.ssl_channel_credentials()
+                envs.ORCA_CORE, grpc.ssl_channel_credentials()
             ) as channel:
                 stub = service_pb2_grpc.OrcaCoreStub(channel)
                 response = stub.RegisterProcessor(registration_request)
                 LOGGER.info(f"Algorithm registration response received: {response}")
         else:
             # insecure channel for local development
-            with grpc.insecure_channel(envs.ORCACORE) as channel:
+            with grpc.insecure_channel(envs.ORCA_CORE) as channel:
                 stub = service_pb2_grpc.OrcaCoreStub(channel)
                 response = stub.RegisterProcessor(registration_request)
                 LOGGER.info(f"Algorithm registration response received: {response}")
@@ -735,7 +731,7 @@ def Start(self) -> None:
             # add the server port
             port = server.add_insecure_port(self._processorConnStr)
             if port == 0:
-                raise RuntimeError(f"Failed to bind to port {envs.PORT}")
+                raise RuntimeError(f"Failed to bind to port {envs.PROCESSOR_PORT}")
 
             LOGGER.info(f"Server listening on address {self._processorConnStr}")
 

pyproject.toml

@@ -61,7 +61,7 @@ markers = [
 [tool.poe.env]
 ORCA_CORE="<dummy_grpc_url>"
 PROCESSOR_PORT="5051"
-PROCESSOR_ADDRESS="[::]"
+PROCESSOR_ADDRESS="[::]:8080"
 
 [tool.poe.tasks]
 _lint_check = "ruff check orca_python tests examples"

tests/test_processor_address.py

@@ -0,0 +1,57 @@
+import pytest
+
+from orca_python.envs import (
+    _parse_connection_string,
+)
+
+
+@pytest.mark.parametrize(
+    "connection_string,expected_address,expected_port",
+    [
+        ("localhost:8080", "localhost", 8080),
+        ("192.168.1.1:443", "192.168.1.1", 443),
+        ("api.service.com:9000", "api.service.com", 9000),
+        ("::1:3000", "::1", 3000),
+        ("127.0.0.1:1", "127.0.0.1", 1),
+        ("multiple:slashes:here:8080", "multiple:slashes:here", 8080),
+    ],
+)
+def test_parse_connection_string_parametrized_valid(
+    connection_string, expected_address, expected_port
+):
+    """Test for valid connection strings"""
+    result = _parse_connection_string(connection_string)
+    assert result == (expected_address, expected_port)
+
+
+@pytest.mark.parametrize(
+    "invalid_connection_string",
+    [
+        "localhost/8080",
+        "localhost:",
+        ":8080",
+        "localhost:abc",
+        "localhost:8080/:extra",
+        "",
+        "no-port",
+        " localhost:8080",
+        "localhost:8080 ",
+    ],
+)
+def test_parse_connection_string_parametrized_invalid(invalid_connection_string):
+    """Test for invalid connection strings"""
+    result = _parse_connection_string(invalid_connection_string)
+    assert result is None
+
+
+def test_parse_connection_string_edge_case_port_zero():
+    """Test parsing with port zero"""
+    result = _parse_connection_string("localhost:0")
+    assert result == ("localhost", 0)
+
+
+def test_parse_connection_string_edge_case_very_long_address():
+    """Test parsing with very long address"""
+    long_address = "a" * 100 + ".example.com"
+    result = _parse_connection_string(f"{long_address}:8080")
+    assert result == (long_address, 8080)