diff --git a/examples/simpletrigger/processor.py b/examples/simpletrigger/processor.py index 630fcdd..65f9793 100644 --- a/examples/simpletrigger/processor.py +++ b/examples/simpletrigger/processor.py @@ -1,16 +1,32 @@ import time -from orca_python import Processor, WindowType, StructResult, ExecutionParams +from orca_python import ( + Processor, + WindowType, + StructResult, + MetadataField, + ExecutionParams, +) proc = Processor("ml") +trip_id = MetadataField(name="trip_id", description="The unique ID of the trip") +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" + name="Every30Second", + version="1.0.0", + description="Triggers every 30 seconds", + metadataFields=[trip_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) + bus_id = params.window.metadata.get("bus_id", None) + print(trip_id, bus_id) + time.sleep(5) return StructResult({"result": 42}) diff --git a/examples/simpletrigger/window.py b/examples/simpletrigger/window.py index eb208fb..d510c55 100644 --- a/examples/simpletrigger/window.py +++ b/examples/simpletrigger/window.py @@ -2,6 +2,7 @@ import datetime as dt import schedule +from processor import bus_id, trip_id from orca_python import Window, EmitWindow @@ -14,6 +15,7 @@ def emitWindow() -> None: name="Every30Second", version="1.0.0", origin="Example", + metadata={bus_id.name: 1, trip_id.name: 2}, ) EmitWindow(window) @@ -21,6 +23,7 @@ def emitWindow() -> None: schedule.every(30).seconds.do(emitWindow) if __name__ == "__main__": + emitWindow() while True: schedule.run_pending() time.sleep(1) diff --git a/orca b/orca index 33369bd..16b6306 160000 --- a/orca +++ b/orca @@ -1 +1 @@ -Subproject commit 33369bd4e916b3583d9e5f8fbaa1f15330a53522 +Subproject commit 16b63068e7011df9622a5fe935eb854ccbe6baac diff --git a/orca_python/__init__.py b/orca_python/__init__.py index e861125..78fb3a5 100644 --- a/orca_python/__init__.py +++ b/orca_python/__init__.py @@ -7,6 +7,7 @@ ArrayResult, ValueResult, StructResult, + MetadataField, ExecutionParams, ) @@ -14,6 +15,7 @@ "Processor", "EmitWindow", "Window", + "MetadataField", "WindowType", "StructResult", "ValueResult", diff --git a/orca_python/envs.py b/orca_python/envs.py index 957ea89..133f636 100644 --- a/orca_python/envs.py +++ b/orca_python/envs.py @@ -27,4 +27,4 @@ def getenvs() -> Tuple[bool, str, str, str]: return is_production, orcaserver, port, host -is_production, ORCASERVER, PORT, HOST = getenvs() +is_production, ORCACORE, PORT, HOST = getenvs() diff --git a/orca_python/exceptions.py b/orca_python/exceptions.py index a9b1544..898f612 100644 --- a/orca_python/exceptions.py +++ b/orca_python/exceptions.py @@ -14,6 +14,10 @@ class InvalidWindowArgument(BaseOrcaException): """Raised when an argument to the Window class is not valid""" +class InvalidMetadataFieldArgument(BaseOrcaException): + """Raised when an argument to a metadata field is not valid""" + + class InvalidDependency(BaseOrcaException): """Raised when a dependency is invalid""" diff --git a/orca_python/main.py b/orca_python/main.py index de8f7b9..3a30f63 100644 --- a/orca_python/main.py +++ b/orca_python/main.py @@ -55,6 +55,7 @@ InvalidWindowArgument, InvalidAlgorithmArgument, InvalidAlgorithmReturnType, + InvalidMetadataFieldArgument, ) # Regex patterns for validation @@ -66,24 +67,49 @@ LOGGER = logging.getLogger(__name__) +@dataclass(frozen=True) +class MetadataField: + name: str + description: str + + def __post_init__(self) -> None: + if self.name == "": + raise InvalidMetadataFieldArgument("Metadata field name cannot be empty") + + if self.description == "": + raise InvalidMetadataFieldArgument( + "Metadata field description cannot be empty" + ) + + @dataclass class WindowType: name: str version: str description: str + metadataFields: List[MetadataField] = field(default_factory=list) - def __post__init__(self) -> None: + def __post_init__(self) -> None: if not re.match(WINDOW_NAME, self.name): - raise InvalidAlgorithmArgument( + raise InvalidWindowArgument( f"Window name '{self.name}' must be in PascalCase" ) if not re.match(SEMVER_PATTERN, self.version): - raise InvalidAlgorithmArgument( + raise InvalidWindowArgument( f"Window version '{self.version}' must follow basic semantic " "versioning (e.g., '1.0.0') without release portions" ) + _seenFields = set() + for field in self.metadataFields: + if field in _seenFields: + raise InvalidWindowArgument( + f"Two or more metadata fields provided with the same name:'{field.name}' and description@ '{field.description}" + ) + else: + _seenFields.add(field) + @dataclass class StructResult: @@ -213,14 +239,14 @@ def EmitWindow(window: Window) -> None: if envs.is_production: # secure channel with TLS with grpc.secure_channel( - envs.ORCASERVER, grpc.ssl_channel_credentials() + envs.ORCACORE, 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.ORCASERVER) as channel: + with grpc.insecure_channel(envs.ORCACORE) as channel: stub = service_pb2_grpc.OrcaCoreStub(channel) response = stub.EmitWindow(window_pb) LOGGER.info(f"Window emitted: {response}") @@ -234,8 +260,7 @@ class Algorithm: Attributes: name (str): The name of the algorithm (PascalCase). version (str): Semantic version of the algorithm (e.g., "1.0.0"). - window_name (str): The window type name that triggers the algorithm. - window_version (str): The version of the window type. + window_type (WindowType): The window type triggers the algorithm. exec_fn (AlgorithmFn): The execution function for the algorithm. processor (str): Name of the processor where it's registered. runtime (str): Python runtime used for execution. @@ -243,9 +268,7 @@ class Algorithm: name: str version: str - window_name: str - window_version: str - window_description: str + window_type: WindowType exec_fn: AlgorithmFn processor: str runtime: str @@ -259,7 +282,7 @@ def full_name(self) -> str: @property def full_window_name(self) -> str: """Returns the full window name as `window_name_window_version`.""" - return f"{self.window_name}_{self.window_version}" + return f"{self.window_type.name}_{self.window_type.version}" class Algorithms: @@ -293,7 +316,7 @@ def _add_algorithm(self, name: str, algorithm: Algorithm) -> None: LOGGER.error(f"Attempted to register duplicate algorithm: {name}") raise ValueError(f"Algorithm {name} already exists") LOGGER.info( - f"Registering algorithm: {name} (window: {algorithm.window_name}_{algorithm.window_version})" + f"Registering algorithm: {name} (window: {algorithm.window_type.name}_{algorithm.window_type.version})" ) self._algorithms[name] = algorithm @@ -641,9 +664,16 @@ def Register(self) -> None: algo_msg.result_type = result_type_pb # Add window type - algo_msg.window_type.name = algorithm.window_name - algo_msg.window_type.version = algorithm.window_version - algo_msg.window_type.description = algorithm.window_description + algo_msg.window_type.name = algorithm.window_type.name + algo_msg.window_type.version = algorithm.window_type.version + algo_msg.window_type.description = algorithm.window_type.description + + # Fill in metadata fields if present + if len(algorithm.window_type.metadataFields) > 0: + for metadataField in algorithm.window_type.metadataFields: + metadata_fields_msg = algo_msg.window_type.metadataFields.add() + metadata_fields_msg.name = metadataField.name + metadata_fields_msg.description = metadataField.description # Add dependencies if they exist if algorithm.full_name in self._algorithmsSingleton._dependencies: @@ -657,14 +687,14 @@ def Register(self) -> None: if envs.is_production: # secure channel with TLS with grpc.secure_channel( - envs.ORCASERVER, grpc.ssl_channel_credentials() + envs.ORCACORE, 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.ORCASERVER) as channel: + with grpc.insecure_channel(envs.ORCACORE) as channel: stub = service_pb2_grpc.OrcaCoreStub(channel) response = stub.RegisterProcessor(registration_request) LOGGER.info(f"Algorithm registration response received: {response}") @@ -804,9 +834,7 @@ def wrapper( algorithm = Algorithm( name=name, version=version, - window_name=window_type.name, - window_version=window_type.version, - window_description=window_type.description, + window_type=window_type, exec_fn=wrapper, processor=self._name, runtime=sys.version, diff --git a/pyproject.toml b/pyproject.toml index f286a73..0bc11a3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,11 @@ markers = [ "live: marks tests as requireing access to live services" ] +[tool.poe.env] +ORCA_CORE="" +PROCESSOR_PORT="5051" +PROCESSOR_ADDRESS="[::]" + [tool.poe.tasks] _lint_check = "ruff check orca_python tests examples" _lint_fix = "ruff check orca_python tests examples --fix " diff --git a/tests/test_windows.py b/tests/test_windows.py new file mode 100644 index 0000000..9000896 --- /dev/null +++ b/tests/test_windows.py @@ -0,0 +1,38 @@ +import pytest + +from orca_python import WindowType, MetadataField +from orca_python.exceptions import InvalidWindowArgument, InvalidMetadataFieldArgument + + +def test_metadata_fields(): + with pytest.raises(InvalidMetadataFieldArgument): + MetadataField(name="", description="test description") + + with pytest.raises(InvalidMetadataFieldArgument): + MetadataField(name="test name", description="") + + with pytest.raises(InvalidMetadataFieldArgument): + MetadataField(name="", description="") + + MetadataField(name="test name", description="test description") + + +def test_window_type_definition(): + with pytest.raises(InvalidWindowArgument): + WindowType( + name="TestWindow", + version="1.0.0", + description="test description", + metadataFields=[ + MetadataField(name="testName", description="test description"), + MetadataField(name="testName", description="test description"), + ], + ) + WindowType( + name="TestWindow", + version="1.0.0", + description="test description", + metadataFields=[ + MetadataField(name="test name", description="test description") + ], + )