diff --git a/CHANGELOG.md b/CHANGELOG.md index 8021c8d..612d586 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/simpletrigger/processor.py b/examples/simpletrigger/processor.py index 65f9793..7d0cfca 100644 --- a/examples/simpletrigger/processor.py +++ b/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}) diff --git a/examples/simpletrigger/window.py b/examples/simpletrigger/window.py index d510c55..3541434 100644 --- a/examples/simpletrigger/window.py +++ b/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) diff --git a/orca_python/envs.py b/orca_python/envs.py index 133f636..510a731 100644 --- a/orca_python/envs.py +++ b/orca_python/envs.py @@ -1,22 +1,54 @@ 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 :" + ) + processor_address, processor_port = res env = os.getenv("ENV", "") if env == "production": @@ -24,7 +56,7 @@ def getenvs() -> Tuple[bool, str, str, str]: 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() diff --git a/orca_python/exceptions.py b/orca_python/exceptions.py index 898f612..e11f804 100644 --- a/orca_python/exceptions.py +++ b/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""" diff --git a/orca_python/main.py b/orca_python/main.py index 3a30f63..cec58c4 100644 --- a/orca_python/main.py +++ b/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}") diff --git a/pyproject.toml b/pyproject.toml index be0dea7..ae5ac66 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -61,7 +61,7 @@ markers = [ [tool.poe.env] ORCA_CORE="" PROCESSOR_PORT="5051" -PROCESSOR_ADDRESS="[::]" +PROCESSOR_ADDRESS="[::]:8080" [tool.poe.tasks] _lint_check = "ruff check orca_python tests examples" diff --git a/tests/test_processor_address.py b/tests/test_processor_address.py new file mode 100644 index 0000000..af50d92 --- /dev/null +++ b/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)