diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 2dc891b830..3af9fed90f 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -5,6 +5,7 @@ This project adheres to [Semantic Versioning](http://semver.org/). ## [Unreleased] +- Added `client.user_defined_functions` for reusable expressions. - Added `client.annotations` for data reviews, phases, and review history. ### What's New diff --git a/python/lib/sift_client/_internal/low_level_wrappers/user_defined_functions.py b/python/lib/sift_client/_internal/low_level_wrappers/user_defined_functions.py new file mode 100644 index 0000000000..09f8f512ee --- /dev/null +++ b/python/lib/sift_client/_internal/low_level_wrappers/user_defined_functions.py @@ -0,0 +1,404 @@ +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING, Any, cast + +from sift.user_defined_functions.v1.user_defined_functions_pb2 import ( + CreateUserDefinedFunctionResponse, + GetUserDefinedFunctionDependentsRequest, + GetUserDefinedFunctionDependentsResponse, + GetUserDefinedFunctionRequest, + GetUserDefinedFunctionResponse, + GetUserDefinedFunctionVersionRequest, + GetUserDefinedFunctionVersionResponse, + GetUserDefinedFunctionVersionsRequest, + GetUserDefinedFunctionVersionsResponse, + ListUserDefinedFunctionsRequest, + ListUserDefinedFunctionsResponse, + ListUserDefinedFunctionVersionsRequest, + ListUserDefinedFunctionVersionsResponse, + UpdateUserDefinedFunctionRequest, + UpdateUserDefinedFunctionResponse, + ValidateUserDefinedFunctionRequest, + ValidateUserDefinedFunctionResponse, +) +from sift.user_defined_functions.v1.user_defined_functions_pb2_grpc import ( + UserDefinedFunctionServiceStub, +) + +from sift_client._internal.low_level_wrappers.base import DEFAULT_PAGE_SIZE, LowLevelClientBase +from sift_client.sift_types.calculated_channel import CalculatedChannel +from sift_client.sift_types.rule import Rule +from sift_client.sift_types.user_defined_function import ( + FunctionDataType, + FunctionInput, + FunctionUsage, + UserDefinedFunction, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, + UserDefinedFunctionValidation, + UserDefinedFunctionVersion, + _function_input_to_proto, +) +from sift_client.transport import WithGrpcClient + +if TYPE_CHECKING: + from sift_client.transport.grpc_transport import GrpcClient + +# Configure logging +logger = logging.getLogger(__name__) + + +class UserDefinedFunctionsLowLevelClient(LowLevelClientBase, WithGrpcClient): + """Low-level client for the UserDefinedFunctionsAPI. + + A thin wrapper around the autogenerated bindings for the UserDefinedFunctionService. + """ + + def __init__(self, grpc_client: GrpcClient): + """Initialize the UserDefinedFunctionsLowLevelClient. + + Args: + grpc_client: The gRPC client to use for making API calls. + """ + super().__init__(grpc_client) + + async def get_function(self, function_id: str) -> UserDefinedFunction: + """Get a function by ID. + + Args: + function_id: The function ID to get. + + Returns: + The UserDefinedFunction. + """ + request = GetUserDefinedFunctionRequest(user_defined_function_id=function_id) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).GetUserDefinedFunction(request) + return UserDefinedFunction._from_proto( + cast("GetUserDefinedFunctionResponse", response).user_defined_function + ) + + async def get_version(self, version_id: str) -> UserDefinedFunctionVersion: + """Get one version of a function. + + Args: + version_id: The version ID to get. + + Returns: + The UserDefinedFunctionVersion. + """ + request = GetUserDefinedFunctionVersionRequest(user_defined_function_version_id=version_id) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).GetUserDefinedFunctionVersion(request) + return UserDefinedFunctionVersion._from_proto( + cast("GetUserDefinedFunctionVersionResponse", response).user_defined_function + ) + + async def get_versions(self, version_ids: list[str]) -> list[UserDefinedFunctionVersion]: + """Get many versions in one call. + + Args: + version_ids: The version IDs to get. + + Returns: + The UserDefinedFunctionVersions. + """ + if not version_ids: + return [] + request = GetUserDefinedFunctionVersionsRequest( + user_defined_function_version_ids=version_ids + ) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).GetUserDefinedFunctionVersions(request) + response = cast("GetUserDefinedFunctionVersionsResponse", response) + return [UserDefinedFunctionVersion._from_proto(f) for f in response.user_defined_functions] + + async def list_functions( + self, + *, + page_size: int | None = DEFAULT_PAGE_SIZE, + page_token: str | None = None, + query_filter: str | None = None, + order_by: str | None = None, + ) -> tuple[list[UserDefinedFunction], str]: + """List one page of functions. + + Args: + page_size: The maximum number of functions to return. + page_token: A page token for pagination. + query_filter: A CEL filter string. + order_by: How to order the retrieved functions. + + Returns: + A tuple of (functions, next_page_token). + """ + request_kwargs: dict[str, Any] = {} + if page_size is not None: + request_kwargs["page_size"] = page_size + if page_token is not None: + request_kwargs["page_token"] = page_token + if query_filter is not None: + request_kwargs["filter"] = query_filter + if order_by is not None: + request_kwargs["order_by"] = order_by + + request = ListUserDefinedFunctionsRequest(**request_kwargs) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).ListUserDefinedFunctions(request) + response = cast("ListUserDefinedFunctionsResponse", response) + + functions = [UserDefinedFunction._from_proto(f) for f in response.user_defined_functions] + return functions, response.next_page_token + + async def list_all_functions( + self, + *, + query_filter: str | None = None, + order_by: str | None = None, + page_size: int | None = DEFAULT_PAGE_SIZE, + max_results: int | None = None, + ) -> list[UserDefinedFunction]: + """List every matching function. + + Args: + query_filter: A CEL filter string. + order_by: How to order the retrieved functions. + page_size: The number of results to fetch per request. + max_results: Maximum number of results to return. + + Returns: + A list of all matching functions. + """ + return await self._handle_pagination( + self.list_functions, + kwargs={"query_filter": query_filter}, + order_by=order_by, + max_results=max_results, + page_size=page_size, + ) + + async def list_versions( + self, + *, + function_id: str | None = None, + name: str | None = None, + page_size: int | None = DEFAULT_PAGE_SIZE, + page_token: str | None = None, + query_filter: str | None = None, + order_by: str | None = None, + ) -> tuple[list[UserDefinedFunctionVersion], str]: + """List one page of versions for a function. + + Args: + function_id: The function whose versions to list. + name: The function name, as an alternative to the ID. + page_size: The maximum number of versions to return. + page_token: A page token for pagination. + query_filter: A CEL filter string. + order_by: How to order the retrieved versions. + + Returns: + A tuple of (versions, next_page_token). + """ + request_kwargs: dict[str, Any] = {} + if function_id is not None: + request_kwargs["user_defined_function_id"] = function_id + if name is not None: + request_kwargs["name"] = name + if page_size is not None: + request_kwargs["page_size"] = page_size + if page_token is not None: + request_kwargs["page_token"] = page_token + if query_filter is not None: + request_kwargs["filter"] = query_filter + if order_by is not None: + request_kwargs["order_by"] = order_by + + request = ListUserDefinedFunctionVersionsRequest(**request_kwargs) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).ListUserDefinedFunctionVersions(request) + response = cast("ListUserDefinedFunctionVersionsResponse", response) + + versions = [ + UserDefinedFunctionVersion._from_proto(f) for f in response.user_defined_functions + ] + return versions, response.next_page_token + + async def list_all_versions( + self, + *, + function_id: str | None = None, + name: str | None = None, + query_filter: str | None = None, + order_by: str | None = None, + page_size: int | None = DEFAULT_PAGE_SIZE, + max_results: int | None = None, + ) -> list[UserDefinedFunctionVersion]: + """List every matching version. + + Args: + function_id: The function whose versions to list. + name: The function name, as an alternative to the ID. + query_filter: A CEL filter string. + order_by: How to order the retrieved versions. + page_size: The number of results to fetch per request. + max_results: Maximum number of results to return. + + Returns: + A list of all matching versions. + """ + return await self._handle_pagination( + self.list_versions, + kwargs={"function_id": function_id, "name": name, "query_filter": query_filter}, + order_by=order_by, + max_results=max_results, + page_size=page_size, + ) + + async def create_function(self, *, create: UserDefinedFunctionCreate) -> UserDefinedFunction: + """Create a function. + + Args: + create: The function definition. + + Returns: + The created UserDefinedFunction. + """ + request_proto = create.to_proto() + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).CreateUserDefinedFunction(request_proto) + return UserDefinedFunction._from_proto( + cast("CreateUserDefinedFunctionResponse", response).user_defined_function + ) + + async def update_function( + self, update: UserDefinedFunctionUpdate, change_notes: str | None = None + ) -> UserDefinedFunction: + """Update select fields of a function. + + Args: + update: The updates to apply. Its `resource_id` must be set. + change_notes: The note to attach to the new version. + + Returns: + The updated UserDefinedFunction. + """ + grpc_function, update_mask = update.to_proto_with_mask() + if change_notes is not None: + grpc_function.user_notes = change_notes + request = UpdateUserDefinedFunctionRequest( + user_defined_function=grpc_function, update_mask=update_mask + ) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).UpdateUserDefinedFunction(request) + return UserDefinedFunction._from_proto( + cast("UpdateUserDefinedFunctionResponse", response).user_defined_function + ) + + async def validate_function( + self, *, expression: str, function_inputs: list[FunctionInput] + ) -> UserDefinedFunctionValidation: + """Check an expression without saving it. + + Args: + expression: The expression to check. + function_inputs: The inputs the expression refers to. + + Returns: + Whether the expression compiles, and its output type or error. + """ + request = ValidateUserDefinedFunctionRequest( + expression=expression, + function_inputs=[ + _function_input_to_proto(**i.model_dump()) for i in function_inputs or [] + ], + ) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).ValidateUserDefinedFunction(request) + response = cast("ValidateUserDefinedFunctionResponse", response) + + if response.WhichOneof("result") == "error": + return UserDefinedFunctionValidation(is_valid=False, error=response.error.error_message) + output_type = response.success.user_defined_function.function_output_type + return UserDefinedFunctionValidation( + is_valid=True, + output_type=FunctionDataType(output_type) if output_type else None, + ) + + async def get_dependents( + self, + *, + function_id: str | None = None, + name: str | None = None, + version_id: str | None = None, + ) -> FunctionUsage: + """Get what depends on a function. + + Exactly one identifier must be provided. + + Args: + function_id: The function ID. + name: The function name. + version_id: A specific version ID. + + Returns: + The functions, calculated channels, and rules that use it. + + Raises: + ValueError: If not exactly one identifier is provided. + """ + given = [v for v in (function_id, name, version_id) if v is not None] + if len(given) != 1: + raise ValueError("Exactly one of function_id, name, or version_id must be provided") + + request_kwargs: dict[str, Any] = {} + if function_id is not None: + request_kwargs["user_defined_function_id"] = function_id + elif name is not None: + request_kwargs["user_defined_function_name"] = name + else: + request_kwargs["user_defined_function_version_id"] = version_id + + request_kwargs["page_size"] = DEFAULT_PAGE_SIZE + request = GetUserDefinedFunctionDependentsRequest(**request_kwargs) + response = await self._grpc_client.get_stub( + UserDefinedFunctionServiceStub + ).GetUserDefinedFunctionDependents(request) + response = cast("GetUserDefinedFunctionDependentsResponse", response) + + # The response pages each set separately but the request takes no page token, + # so a leftover token cannot be followed. Say so rather than return short. + truncated = [ + name + for name, token in ( + ("functions", response.next_page_token_user_defined_function), + ("calculated_channels", response.next_page_token_calculated_channel), + ("rules", response.next_page_token_rule), + ) + if token + ] + if truncated: + logger.warning( + "GetUserDefinedFunctionDependents returned more than %d %s; " + "the request takes no page token, so the rest cannot be fetched.", + DEFAULT_PAGE_SIZE, + " and ".join(truncated), + ) + + return FunctionUsage( + functions=[UserDefinedFunction._from_proto(f) for f in response.user_defined_functions], + calculated_channels=[ + CalculatedChannel._from_proto(c) for c in response.calculated_channels + ], + rules=[Rule._from_proto(r) for r in response.rules], + truncated=bool(truncated), + ) diff --git a/python/lib/sift_client/_tests/conftest.py b/python/lib/sift_client/_tests/conftest.py index 28d3ee11d2..e98f30e2d9 100644 --- a/python/lib/sift_client/_tests/conftest.py +++ b/python/lib/sift_client/_tests/conftest.py @@ -93,6 +93,7 @@ def mock_client(): client.jobs = MagicMock() client.data_import = MagicMock() client.data_export = MagicMock() + client.user_defined_functions = MagicMock() client.async_ = MagicMock(spec=AsyncAPIs) client.async_.ingestion = MagicMock() return client diff --git a/python/lib/sift_client/_tests/resources/test_user_defined_functions.py b/python/lib/sift_client/_tests/resources/test_user_defined_functions.py new file mode 100644 index 0000000000..5e9a89c061 --- /dev/null +++ b/python/lib/sift_client/_tests/resources/test_user_defined_functions.py @@ -0,0 +1,332 @@ +"""Pytest tests for the User Defined Functions API. + +These tests demonstrate and validate the usage of the UserDefinedFunctionsAPI including: +- Basic function operations (get, list, find) +- Creation, updates, and archiving +- Versions, validation, dependents, and sync +""" + +from datetime import datetime, timezone + +import pytest + +from sift_client import SiftClient +from sift_client.resources import ( + UserDefinedFunctionsAPI, + UserDefinedFunctionsAPIAsync, + UserDefinedFunctionVersionsAPI, +) +from sift_client.sift_types import UserDefinedFunction +from sift_client.sift_types.user_defined_function import ( + FunctionDataType, + FunctionInput, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, +) + +pytestmark = pytest.mark.integration + + +def test_client_binding(sift_client): + assert isinstance(sift_client.user_defined_functions, UserDefinedFunctionsAPI) + assert isinstance(sift_client.user_defined_functions.versions, UserDefinedFunctionVersionsAPI) + assert isinstance(sift_client.async_.user_defined_functions, UserDefinedFunctionsAPIAsync) + + +@pytest.fixture(scope="session") +def test_timestamp_str(): + """A per-session suffix so function names stay unique across runs.""" + return datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + + +@pytest.fixture +def functions_api_async(sift_client: SiftClient): + """Get the async functions API instance.""" + return sift_client.async_.user_defined_functions + + +@pytest.fixture(scope="session") +def new_function(sift_client, test_timestamp_str): + """Create a function for the session and archive it on teardown.""" + created = sift_client.user_defined_functions.create( + UserDefinedFunctionCreate( + name=f"test_fn_{test_timestamp_str}", + description="Created by Sift Client pytest", + expression="$1 * 2", + function_inputs=[FunctionInput(identifier="$1")], + ) + ) + yield created + sift_client.user_defined_functions.archive(created) + + +class TestUserDefinedFunctions: + """Tests for the User Defined Functions API.""" + + def test_create(self, new_function, test_timestamp_str): + """Test that create returns a fully populated function.""" + assert isinstance(new_function, UserDefinedFunction) + assert new_function.id_ is not None + assert new_function.name == f"test_fn_{test_timestamp_str}" + assert new_function.expression == "$1 * 2" + assert [i.identifier for i in new_function.function_inputs] == ["$1"] + assert new_function.version == 1 + assert new_function.version_id + assert new_function.is_archived is False + + def test_get(self, sift_client, new_function): + """Test getting a function by ID.""" + fetched = sift_client.user_defined_functions.get(new_function._id_or_error) + + assert fetched.id_ == new_function.id_ + assert fetched.name == new_function.name + + def test_basic_list(self, sift_client, new_function): + """Test basic function listing functionality.""" + functions = sift_client.user_defined_functions.list_(limit=5) + + assert isinstance(functions, list) + for function in functions: + assert isinstance(function, UserDefinedFunction) + assert function.id_ is not None + + def test_list_with_name_filter(self, sift_client, new_function): + """Test function listing with name filtering.""" + by_name = sift_client.user_defined_functions.list_(name=new_function.name) + + assert [f.id_ for f in by_name] == [new_function.id_] + + def test_list_with_id_filter(self, sift_client, new_function): + """Test function listing filtered to specific IDs.""" + functions = sift_client.user_defined_functions.list_( + user_defined_function_ids=[new_function] + ) + + assert [f.id_ for f in functions] == [new_function.id_] + + def test_find(self, sift_client, new_function): + """Test finding a single function.""" + found = sift_client.user_defined_functions.find(name=new_function.name) + + assert found is not None + assert found.id_ == new_function.id_ + + def test_find_nonexistent(self, sift_client): + """Test finding a non-existent function returns None.""" + found = sift_client.user_defined_functions.find( + name=f"nonexistent_fn_{datetime.now(timezone.utc).timestamp()}" + ) + assert found is None + + def test_update_creates_a_version(self, sift_client, new_function): + """Test that updating a function bumps its version.""" + updated = sift_client.user_defined_functions.update( + new_function, UserDefinedFunctionUpdate(description="tweaked") + ) + + assert updated.description == "tweaked" + assert updated.version > new_function.version + # The name was not in the mask, so it is unchanged. + assert updated.name == new_function.name + + def test_update_accepts_dict(self, sift_client, new_function): + """Test that update accepts a plain dict.""" + updated = sift_client.user_defined_functions.update( + new_function._id_or_error, {"description": "via dict"} + ) + + assert updated.description == "via dict" + + def test_archive_and_unarchive(self, sift_client, test_timestamp_str): + """Test archiving and unarchiving a function.""" + function = sift_client.user_defined_functions.create( + UserDefinedFunctionCreate( + name=f"test_fn_archive_{test_timestamp_str}", + expression="$1 + 1", + function_inputs=[FunctionInput(identifier="$1")], + ) + ) + + archived = sift_client.user_defined_functions.archive(function) + assert archived.is_archived is True + + # Archived functions are excluded from list_ by default. + assert sift_client.user_defined_functions.find(name=function.name) is None + assert ( + sift_client.user_defined_functions.find(name=function.name, include_archived=True).id_ + == function.id_ + ) + + unarchived = sift_client.user_defined_functions.unarchive(function) + assert unarchived.is_archived is False + + sift_client.user_defined_functions.archive(function) + + def test_instance_methods(self, sift_client, test_timestamp_str): + """Test the update and archive methods on the instance itself.""" + function = sift_client.user_defined_functions.create( + UserDefinedFunctionCreate( + name=f"test_fn_instance_{test_timestamp_str}", + expression="$1 - 1", + function_inputs=[FunctionInput(identifier="$1")], + ) + ) + + function.update({"description": "from-instance"}) + assert function.description == "from-instance" + + function.archive() + assert function.is_archived is True + + def test_validate_accepts_a_good_expression(self, sift_client): + """Test validating an expression that compiles.""" + result = sift_client.user_defined_functions.validate_expression( + "$1 * 2", [FunctionInput(identifier="$1")] + ) + + assert result.valid is True + assert result.error is None + + def test_validate_rejects_a_bad_expression(self, sift_client): + """Test validating an expression that does not compile.""" + result = sift_client.user_defined_functions.validate_expression( + "$1 ** ** 2", [FunctionInput(identifier="$1")] + ) + + assert result.valid is False + assert result.error + + def test_dependents_of_an_unused_function(self, sift_client, new_function): + """Test that a function nothing uses reports no dependents.""" + dependents = sift_client.user_defined_functions.get_where_used(new_function) + + assert dependents.is_used is False + + def test_dependents_come_back_usable(self, sift_client, new_function, test_timestamp_str): + """Test that a returned object carries the client, so its properties resolve.""" + from sift_client.sift_types.calculated_channel import ( + CalculatedChannelCreate, + ChannelReference, + ) + + channel = sift_client.channels.list_(limit=1)[0] + calc = sift_client.calculated_channels.create( + CalculatedChannelCreate( + name=f"test_fn_user_{test_timestamp_str}", + expression=f"{new_function.name}($1)", + expression_channel_references=[ + ChannelReference(channel_reference="$1", channel_identifier=channel.name) + ], + all_assets=True, + ) + ) + + usage = sift_client.user_defined_functions.get_where_used(new_function) + + assert [c.id_ for c in usage.calculated_channels] == [calc.id_] + # CalculatedChannel exposes no client-backed property, so check the client + # directly. Without it, every property on a returned object raises. + assert usage.calculated_channels[0]._client is not None + + sift_client.calculated_channels.archive(calc) + + @pytest.mark.asyncio + async def test_async_list(self, functions_api_async, new_function): + """Test the async API returns the same functions.""" + functions = await functions_api_async.list_(user_defined_function_ids=[new_function]) + + assert [f.id_ for f in functions] == [new_function.id_] + + +class TestUserDefinedFunctionVersions: + """Tests for the nested versions API.""" + + def test_list_versions(self, sift_client, new_function): + """Test listing a function's versions.""" + versions = sift_client.user_defined_functions.versions.list_( + user_defined_function=new_function + ) + + assert versions + assert all(v.id_ == new_function.id_ for v in versions) + + def test_get_version(self, sift_client, new_function): + """Test getting one version by ID.""" + version = sift_client.user_defined_functions.versions.get(new_function.version_id) + + assert version.version_id == new_function.version_id + + def test_versions_property(self, sift_client, new_function): + """Test the versions property on the instance.""" + assert new_function.versions + + +class TestUserDefinedFunctionSync: + """Tests for sync, which creates or updates by name.""" + + def test_sync_creates_then_updates(self, sift_client, test_timestamp_str): + """Test that sync creates a missing function and updates an existing one.""" + name = f"test_fn_sync_{test_timestamp_str}" + + created = sift_client.user_defined_functions.sync( + [ + UserDefinedFunctionCreate( + name=name, + expression="$1 * 2", + function_inputs=[FunctionInput(identifier="$1")], + ) + ] + ) + assert len(created) == 1 + assert created[0].name == name + first_version = created[0].version + + updated = sift_client.user_defined_functions.sync( + [ + UserDefinedFunctionCreate( + name=name, + expression="$1 * 2", + function_inputs=[FunctionInput(identifier="$1")], + description="second pass", + ) + ] + ) + assert updated[0].id_ == created[0].id_ + assert updated[0].version > first_version + assert updated[0].description == "second pass" + + sift_client.user_defined_functions.archive(updated[0]) + + def test_sync_is_ordered(self, sift_client, test_timestamp_str): + """Test that sync returns results in the order given.""" + names = [f"test_fn_sync{i}_{test_timestamp_str}" for i in range(2)] + + results = sift_client.user_defined_functions.sync( + [ + UserDefinedFunctionCreate( + name=n, expression="$1 + 1", function_inputs=[FunctionInput(identifier="$1")] + ) + for n in names + ] + ) + + assert [r.name for r in results] == names + for result in results: + sift_client.user_defined_functions.archive(result) + + +class TestFunctionDataTypes: + """Tests that non-numeric input types round trip.""" + + def test_string_input(self, sift_client, test_timestamp_str): + """Test a function with a string input.""" + function = sift_client.user_defined_functions.create( + UserDefinedFunctionCreate( + name=f"test_fn_string_{test_timestamp_str}", + expression="$1", + function_inputs=[FunctionInput(identifier="$1", data_type=FunctionDataType.STRING)], + ) + ) + + assert function.function_inputs[0].data_type is FunctionDataType.STRING + sift_client.user_defined_functions.archive(function) diff --git a/python/lib/sift_client/_tests/sift_types/test_user_defined_function.py b/python/lib/sift_client/_tests/sift_types/test_user_defined_function.py new file mode 100644 index 0000000000..9701b02cb6 --- /dev/null +++ b/python/lib/sift_client/_tests/sift_types/test_user_defined_function.py @@ -0,0 +1,203 @@ +"""Tests for sift_types.UserDefinedFunction model.""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock + +import pytest + +from sift_client.sift_types import UserDefinedFunction +from sift_client.sift_types.user_defined_function import ( + FunctionDataType, + FunctionInput, + FunctionUsage, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, + UserDefinedFunctionVersion, +) + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +class TestFunctionInput: + """Unit tests for FunctionInput.""" + + def test_defaults(self): + given = FunctionInput(identifier="$1") + + assert given.data_type is FunctionDataType.NUMERIC + assert given.scalar is False + + def test_round_trip(self): + from sift.common.type.v1.user_defined_functions_pb2 import FunctionInput as Proto + + original = FunctionInput(identifier="$1", data_type=FunctionDataType.STRING, scalar=True) + proto = Proto(identifier="$1", data_type=FunctionDataType.STRING.value, constant=True) + + assert FunctionInput._from_proto(proto) == original + + +class TestUserDefinedFunctionCreate: + """Unit tests for UserDefinedFunctionCreate - tests _to_proto_helpers.""" + + def test_minimal_create(self): + proto = UserDefinedFunctionCreate( + name="double", expression="$1 * 2", function_inputs=[FunctionInput(identifier="$1")] + ).to_proto() + + assert proto.name == "double" + assert proto.expression == "$1 * 2" + + def test_function_inputs_converter(self): + proto = UserDefinedFunctionCreate( + name="double", + expression="$1 * 2", + function_inputs=[ + FunctionInput(identifier="$1"), + FunctionInput(identifier="$2", data_type=FunctionDataType.STRING, scalar=True), + ], + ).to_proto() + + assert len(proto.function_inputs) == 2 + assert proto.function_inputs[0].identifier == "$1" + assert proto.function_inputs[0].data_type == FunctionDataType.NUMERIC.value + assert proto.function_inputs[1].data_type == FunctionDataType.STRING.value + assert proto.function_inputs[1].constant is True + + def test_metadata_converter(self): + proto = UserDefinedFunctionCreate( + name="double", + expression="$1 * 2", + function_inputs=[FunctionInput(identifier="$1")], + metadata={"owner": "ops", "n": 2.0}, + ).to_proto() + + by_key = {m.key.name: m for m in proto.metadata} + assert by_key["owner"].string_value == "ops" + assert by_key["n"].number_value == 2.0 + + +class TestUserDefinedFunctionUpdate: + """Unit tests for UserDefinedFunctionUpdate - tests field masks.""" + + def test_update_mask_only_includes_set_fields(self): + update = UserDefinedFunctionUpdate(expression="$1 * 3", description="tripled") + update.resource_id = "fn-1" + + proto, mask = update.to_proto_with_mask() + + assert proto.user_defined_function_id == "fn-1" + assert proto.expression == "$1 * 3" + assert set(mask.paths) == {"expression", "description"} + + def test_archive_update(self): + update = UserDefinedFunctionUpdate(is_archived=True) + update.resource_id = "fn-1" + + proto, mask = update.to_proto_with_mask() + + assert proto.is_archived is True + assert mask.paths == ["is_archived"] + + def test_requires_resource_id(self): + with pytest.raises(ValueError, match="Resource ID must be set"): + UserDefinedFunctionUpdate(expression="$1").to_proto_with_mask() + + +class TestFunctionUsage: + """Unit tests for FunctionUsage.""" + + def test_any_is_false_when_empty(self): + assert FunctionUsage().is_used is False + + def test_any_is_true_with_one_rule(self): + usage = FunctionUsage.model_construct(rules=[object()]) + + assert usage.is_used is True + + +@pytest.fixture +def mock_function(mock_client): + """Create a mock UserDefinedFunction instance for testing.""" + latest = UserDefinedFunctionVersion( + proto=MagicMock(), + id_="ver-1", + user_defined_function_id="fn-1", + version=1, + expression="$1 * 2", + change_message="created", + change_notes="", + function_inputs=[FunctionInput(identifier="$1")], + dependency_version_ids=[], + created_date=NOW, + created_by_user_id="user1", + output_type=FunctionDataType.NUMERIC, + ) + function = UserDefinedFunction( + proto=MagicMock(), + id_="fn-1", + name="double", + description="doubles the input", + metadata={}, + created_date=NOW, + modified_date=NOW, + created_by_user_id="user1", + modified_by_user_id="user1", + is_archived=False, + latest_version=latest, + archived_date=None, + ) + function._apply_client_to_instance(mock_client) + return function + + +class TestUserDefinedFunction: + """Unit tests for UserDefinedFunction model - tests properties and methods.""" + + def test_versions_property_calls_client(self, mock_function, mock_client): + mock_client.user_defined_functions.versions.list_.return_value = [] + + _ = mock_function.versions + + mock_client.user_defined_functions.versions.list_.assert_called_once_with( + user_defined_function=mock_function, order_by="version desc" + ) + + def test_update_calls_client_and_updates_self(self, mock_function, mock_client): + updated = MagicMock() + mock_client.user_defined_functions.update.return_value = updated + + with MagicMock() as mock_update: + mock_function._update = mock_update + + update = UserDefinedFunctionUpdate(expression="$1 * 3") + result = mock_function.update(update) + + mock_client.user_defined_functions.update.assert_called_once_with( + user_defined_function=mock_function, update=update + ) + mock_update.assert_called_once_with(updated) + assert result is mock_function + + def test_archive_calls_client(self, mock_function, mock_client): + mock_client.user_defined_functions.archive.return_value = MagicMock() + + with MagicMock() as mock_update: + mock_function._update = mock_update + result = mock_function.archive() + + mock_client.user_defined_functions.archive.assert_called_once_with( + user_defined_function=mock_function + ) + assert result is mock_function + + def test_unarchive_calls_client(self, mock_function, mock_client): + mock_client.user_defined_functions.unarchive.return_value = MagicMock() + + with MagicMock() as mock_update: + mock_function._update = mock_update + result = mock_function.unarchive() + + mock_client.user_defined_functions.unarchive.assert_called_once_with( + user_defined_function=mock_function + ) + assert result is mock_function diff --git a/python/lib/sift_client/client.py b/python/lib/sift_client/client.py index dc1a46e8ba..c6530ac77a 100644 --- a/python/lib/sift_client/client.py +++ b/python/lib/sift_client/client.py @@ -42,6 +42,8 @@ TagsAPIAsync, TestResultsAPI, TestResultsAPIAsync, + UserDefinedFunctionsAPI, + UserDefinedFunctionsAPIAsync, UsersAPI, UsersAPIAsync, ) @@ -142,6 +144,9 @@ class SiftClient( test_results: TestResultsAPI """Instance of the Test Results API for making synchronous requests.""" + user_defined_functions: UserDefinedFunctionsAPI + """Instance of the User Defined Functions API for making synchronous requests.""" + users: UsersAPI """Instance of the Users API for making synchronous requests.""" @@ -254,6 +259,7 @@ def __init__( ) self.tags = TagsAPI(self) self.test_results = TestResultsAPI(self) + self.user_defined_functions = UserDefinedFunctionsAPI(self) self.users = UsersAPI(self) self.data_export = DataExportAPI(self) self.data_import = DataImportAPI(self) @@ -277,6 +283,7 @@ def __init__( ), tags=TagsAPIAsync(self), test_results=TestResultsAPIAsync(self), + user_defined_functions=UserDefinedFunctionsAPIAsync(self), users=UsersAPIAsync(self), data_export=DataExportAPIAsync(self), data_import=DataImportAPIAsync(self), diff --git a/python/lib/sift_client/resources/__init__.py b/python/lib/sift_client/resources/__init__.py index 99861cd669..bf180462d5 100644 --- a/python/lib/sift_client/resources/__init__.py +++ b/python/lib/sift_client/resources/__init__.py @@ -181,6 +181,10 @@ async def main(): from sift_client.resources.runs import RunsAPIAsync from sift_client.resources.tags import TagsAPIAsync from sift_client.resources.test_results import TestResultsAPIAsync +from sift_client.resources.user_defined_functions import ( + UserDefinedFunctionsAPIAsync, + UserDefinedFunctionVersionsAPIAsync, +) from sift_client.resources.users import UsersAPIAsync from sift_client.resources.data_imports import DataImportAPIAsync from sift_client.resources.exports import DataExportAPIAsync @@ -208,6 +212,8 @@ async def main(): RunsAPI, TagsAPI, TestResultsAPI, + UserDefinedFunctionsAPI, + UserDefinedFunctionVersionsAPI, UsersAPI, FileAttachmentsAPI, DataExportAPI, @@ -272,6 +278,10 @@ async def main(): "TestResultsAPI", "TestResultsAPIAsync", "TracingConfig", + "UserDefinedFunctionVersionsAPI", + "UserDefinedFunctionVersionsAPIAsync", + "UserDefinedFunctionsAPI", + "UserDefinedFunctionsAPIAsync", "UsersAPI", "UsersAPIAsync", "DataExportAPI", diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.py b/python/lib/sift_client/resources/sync_stubs/__init__.py index 64206b780c..1c59f30589 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.py +++ b/python/lib/sift_client/resources/sync_stubs/__init__.py @@ -28,6 +28,8 @@ RunsAPIAsync, TagsAPIAsync, TestResultsAPIAsync, + UserDefinedFunctionsAPIAsync, + UserDefinedFunctionVersionsAPIAsync, UsersAPIAsync, ) @@ -89,6 +91,15 @@ ) TagsAPI = generate_sync_api(TagsAPIAsync, "TagsAPI") TestResultsAPI = generate_sync_api(TestResultsAPIAsync, "TestResultsAPI") +# UserDefinedFunctionVersionsAPI must be generated before UserDefinedFunctionsAPI so it can be nested under it. +UserDefinedFunctionVersionsAPI = generate_sync_api( + UserDefinedFunctionVersionsAPIAsync, "UserDefinedFunctionVersionsAPI" +) +UserDefinedFunctionsAPI = generate_sync_api( + UserDefinedFunctionsAPIAsync, + "UserDefinedFunctionsAPI", + nested_resources={"versions": UserDefinedFunctionVersionsAPI}, +) UsersAPI = generate_sync_api(UsersAPIAsync, "UsersAPI") DataExportAPI = generate_sync_api(DataExportAPIAsync, "DataExportAPI") DataImportAPI = generate_sync_api(DataImportAPIAsync, "DataImportAPI") @@ -118,5 +129,7 @@ "RunsAPI", "TagsAPI", "TestResultsAPI", + "UserDefinedFunctionVersionsAPI", + "UserDefinedFunctionsAPI", "UsersAPI", ] diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index 8275d1c63a..beb9585a81 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -96,6 +96,15 @@ if TYPE_CHECKING: TestStepUpdate, ) from sift_client.sift_types.user import User + from sift_client.sift_types.user_defined_function import ( + FunctionInput, + FunctionUsage, + UserDefinedFunction, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, + UserDefinedFunctionValidation, + UserDefinedFunctionVersion, + ) class AnnotationLogsAPI: """Sync counterpart to `AnnotationLogsAPIAsync`. @@ -3778,6 +3787,253 @@ class TestResultsAPI: """ ... +class UserDefinedFunctionVersionsAPI: + """Sync counterpart to `UserDefinedFunctionVersionsAPIAsync`. + + High-level API for a function's version history. + + Every save produces a new version. Reachable as + `client.user_defined_functions.versions`. + """ + + def __init__(self, sift_client: SiftClient): + """Initialize the UserDefinedFunctionVersionsAPI. + + Args: + sift_client: The Sift client to use. + """ + ... + + def _run(self, coro): ... + def batch_get( + self, *, versions: list[str] | list[UserDefinedFunctionVersion] + ) -> list[UserDefinedFunctionVersion]: + """Get many versions in one call. + + Args: + versions: The UserDefinedFunctionVersions or version IDs. + + Returns: + The UserDefinedFunctionVersions. + """ + ... + + def get(self, *, version: str | UserDefinedFunctionVersion) -> UserDefinedFunctionVersion: + """Get one version. + + Args: + version: The UserDefinedFunctionVersion or version ID. + + Returns: + The UserDefinedFunctionVersion. + """ + ... + + def list_( + self, + *, + user_defined_function: str | UserDefinedFunction | None = None, + name: str | None = None, + version: int | None = None, + include_archived: bool = False, + filter_query: str | None = None, + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[UserDefinedFunctionVersion]: + """List a function's versions. + + Args: + user_defined_function: The UserDefinedFunction or function ID whose versions to list. + name: The function name, as an alternative to `user_defined_function`. + version: Filter to a single version number. + include_archived: If True, include archived versions in results. + filter_query: Explicit CEL query to filter versions. + order_by: Field and direction to order results by. + limit: Maximum number of versions to return. If None, returns all matches. + page_size: Number of results to fetch per request. + + Returns: + A list of UserDefinedFunctionVersion objects in the service's order, which is + by name unless `order_by` says otherwise. Every version shares the function's + name, so pass `order_by="version desc"` for newest first. + """ + ... + +class UserDefinedFunctionsAPI: + """Sync counterpart to `UserDefinedFunctionsAPIAsync`. + + High-level API for interacting with user-defined functions. + + A user-defined function is a named expression that calculated channels and rules + can call. Every save produces a new version. + """ + + def __init__(self, sift_client: SiftClient): + """Initialize the UserDefinedFunctionsAPI. + + Args: + sift_client: The Sift client to use. + """ + ... + + def _run(self, coro): ... + def archive(self, user_defined_function: str | UserDefinedFunction) -> UserDefinedFunction: + """Archive a function. + + Args: + user_defined_function: The UserDefinedFunction or function ID to archive. + + Returns: + The archived UserDefinedFunction. + """ + ... + + def create(self, create: UserDefinedFunctionCreate | dict) -> UserDefinedFunction: + """Create a new function. + + Args: + create: The function definition. + + Returns: + The created UserDefinedFunction. + """ + ... + + def find(self, **kwargs) -> UserDefinedFunction | None: + """Find one function. Takes the same arguments as `list_`. + + Raises if more than one matches. + + Args: + **kwargs: Keyword arguments to pass to `list_`. + + Returns: + The UserDefinedFunction found or None. + """ + ... + + def get(self, *, user_defined_function_id: str) -> UserDefinedFunction: + """Get a UserDefinedFunction. + + Args: + user_defined_function_id: The ID of the function. + + Returns: + The UserDefinedFunction. + """ + ... + + def get_where_used( + self, + user_defined_function: str | UserDefinedFunction | None = None, + *, + version: str | UserDefinedFunctionVersion | None = None, + ) -> FunctionUsage: + """Get what uses a function. + + Check this before changing inputs or the output type. The server refuses those + changes once a function is in use. + + Each result set is capped at 1000. The service pages them separately but accepts + no page token, so the rest cannot be fetched; `FunctionUsage.truncated` says when + that happened. + + Args: + user_defined_function: The UserDefinedFunction or function ID. + version: A specific UserDefinedFunctionVersion or version ID, instead of the + function as a whole. + + Returns: + The functions, calculated channels, and rules that use it. + + Raises: + ValueError: If neither or both are provided. + """ + ... + + def list_( + self, + *, + name: str | None = None, + name_contains: str | None = None, + name_regex: str | re.Pattern | None = None, + user_defined_function_ids: list[str] | list[UserDefinedFunction] | None = None, + include_archived: bool = False, + filter_query: str | None = None, + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[UserDefinedFunction]: + """List user-defined functions. + + Args: + name: Exact name of the function. + name_contains: Partial name of the function. + name_regex: Regular expression to filter functions by name. + user_defined_function_ids: Filter to these UserDefinedFunctions or function IDs. + include_archived: If True, include archived functions in results. + filter_query: Explicit CEL query to filter functions. + order_by: Field and direction to order results by. + limit: Maximum number of functions to return. If None, returns all matches. + page_size: Number of results to fetch per request. + + Returns: + A list of UserDefinedFunction objects that match the filter criteria. + """ + ... + + def unarchive(self, user_defined_function: str | UserDefinedFunction) -> UserDefinedFunction: + """Unarchive a function. + + Args: + user_defined_function: The UserDefinedFunction or function ID to unarchive. + + Returns: + The unarchived UserDefinedFunction. + """ + ... + + def update( + self, + user_defined_function: str | UserDefinedFunction, + update: UserDefinedFunctionUpdate | dict, + change_notes: str | None = None, + ) -> UserDefinedFunction: + """Update a function. + + Changes to the expression, inputs, description, or metadata create a new version. + + Args: + user_defined_function: The UserDefinedFunction or function ID to update. + update: Updates to apply to the function. + change_notes: A note to attach to the new version. The server treats an empty + note as a change, so the current version's note carries over when omitted. + + Returns: + The updated UserDefinedFunction. + """ + ... + + def validate_expression( + self, expression: str, function_inputs: list[FunctionInput] + ) -> UserDefinedFunctionValidation: + """Check an expression without saving it. + + Args: + expression: The expression to check. + function_inputs: The inputs the expression refers to. The server rejects an + empty list. + + Returns: + Whether the expression compiles, and its output type or error. + """ + ... + @property + def versions(self) -> UserDefinedFunctionVersionsAPI: + """Nested UserDefinedFunctionVersionsAPI for making synchronous requests.""" + ... + class UsersAPI: """Sync counterpart to `UsersAPIAsync`. diff --git a/python/lib/sift_client/resources/user_defined_functions.py b/python/lib/sift_client/resources/user_defined_functions.py new file mode 100644 index 0000000000..d3ace3c951 --- /dev/null +++ b/python/lib/sift_client/resources/user_defined_functions.py @@ -0,0 +1,371 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from sift_client._internal.low_level_wrappers.user_defined_functions import ( + UserDefinedFunctionsLowLevelClient, +) +from sift_client.resources._base import ResourceBase +from sift_client.sift_types.user_defined_function import ( + FunctionInput, + FunctionUsage, + UserDefinedFunction, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, + UserDefinedFunctionValidation, + UserDefinedFunctionVersion, +) +from sift_client.util import cel_utils as cel + +if TYPE_CHECKING: + import re + + from sift_client.client import SiftClient + + +class UserDefinedFunctionVersionsAPIAsync(ResourceBase): + """High-level API for a function's version history. + + Every save produces a new version. Reachable as + `client.user_defined_functions.versions`. + """ + + def __init__(self, sift_client: SiftClient): + """Initialize the UserDefinedFunctionVersionsAPI. + + Args: + sift_client: The Sift client to use. + """ + super().__init__(sift_client) + self._low_level_client = UserDefinedFunctionsLowLevelClient( + grpc_client=self.client.grpc_client + ) + + async def get(self, *, version: str | UserDefinedFunctionVersion) -> UserDefinedFunctionVersion: + """Get one version. + + Args: + version: The UserDefinedFunctionVersion or version ID. + + Returns: + The UserDefinedFunctionVersion. + """ + version_id = ( + version._id_or_error if isinstance(version, UserDefinedFunctionVersion) else version + ) + version = await self._low_level_client.get_version(version_id=version_id) + return self._apply_client_to_instance(version) + + async def batch_get( + self, *, versions: list[str] | list[UserDefinedFunctionVersion] + ) -> list[UserDefinedFunctionVersion]: + """Get many versions in one call. + + Args: + versions: The UserDefinedFunctionVersions or version IDs. + + Returns: + The UserDefinedFunctionVersions. + """ + ids = [v._id_or_error if isinstance(v, UserDefinedFunctionVersion) else v for v in versions] + found = await self._low_level_client.get_versions(version_ids=ids) + return self._apply_client_to_instances(found) + + async def list_( + self, + *, + user_defined_function: str | UserDefinedFunction | None = None, + name: str | None = None, + # version specific + version: int | None = None, + # common filters + include_archived: bool = False, + filter_query: str | None = None, + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[UserDefinedFunctionVersion]: + """List a function's versions. + + Args: + user_defined_function: The UserDefinedFunction or function ID whose versions to list. + name: The function name, as an alternative to `user_defined_function`. + version: Filter to a single version number. + include_archived: If True, include archived versions in results. + filter_query: Explicit CEL query to filter versions. + order_by: Field and direction to order results by. + limit: Maximum number of versions to return. If None, returns all matches. + page_size: Number of results to fetch per request. + + Returns: + A list of UserDefinedFunctionVersion objects in the service's order, which is + by name unless `order_by` says otherwise. Every version shares the function's + name, so pass `order_by="version desc"` for newest first. + """ + filter_parts = self._build_common_cel_filters( + include_archived=include_archived, filter_query=filter_query + ) + if version is not None: + filter_parts.append(cel.equals("version", version)) + query_filter = cel.and_(*filter_parts) + + versions = await self._low_level_client.list_all_versions( + function_id=( + user_defined_function._id_or_error + if isinstance(user_defined_function, UserDefinedFunction) + else user_defined_function + ), + name=name, + query_filter=query_filter or None, + order_by=order_by, + max_results=limit, + **({"page_size": page_size} if page_size is not None else {}), + ) + return self._apply_client_to_instances(versions) + + +class UserDefinedFunctionsAPIAsync(ResourceBase): + """High-level API for interacting with user-defined functions. + + A user-defined function is a named expression that calculated channels and rules + can call. Every save produces a new version. + """ + + def __init__(self, sift_client: SiftClient): + """Initialize the UserDefinedFunctionsAPI. + + Args: + sift_client: The Sift client to use. + """ + super().__init__(sift_client) + self._low_level_client = UserDefinedFunctionsLowLevelClient( + grpc_client=self.client.grpc_client + ) + self.versions = UserDefinedFunctionVersionsAPIAsync(sift_client) + + async def get(self, *, user_defined_function_id: str) -> UserDefinedFunction: + """Get a UserDefinedFunction. + + Args: + user_defined_function_id: The ID of the function. + + Returns: + The UserDefinedFunction. + """ + function = await self._low_level_client.get_function(function_id=user_defined_function_id) + return self._apply_client_to_instance(function) + + async def list_( + self, + *, + name: str | None = None, + name_contains: str | None = None, + name_regex: str | re.Pattern | None = None, + # self ids + user_defined_function_ids: list[str] | list[UserDefinedFunction] | None = None, + # common filters + include_archived: bool = False, + filter_query: str | None = None, + order_by: str | None = None, + limit: int | None = None, + page_size: int | None = None, + ) -> list[UserDefinedFunction]: + """List user-defined functions. + + Args: + name: Exact name of the function. + name_contains: Partial name of the function. + name_regex: Regular expression to filter functions by name. + user_defined_function_ids: Filter to these UserDefinedFunctions or function IDs. + include_archived: If True, include archived functions in results. + filter_query: Explicit CEL query to filter functions. + order_by: Field and direction to order results by. + limit: Maximum number of functions to return. If None, returns all matches. + page_size: Number of results to fetch per request. + + Returns: + A list of UserDefinedFunction objects that match the filter criteria. + """ + filter_parts = [ + *self._build_name_cel_filters( + name=name, name_contains=name_contains, name_regex=name_regex + ), + *self._build_common_cel_filters( + include_archived=include_archived, filter_query=filter_query + ), + ] + if user_defined_function_ids: + ids = [ + f._id_or_error if isinstance(f, UserDefinedFunction) else f + for f in user_defined_function_ids + ] + filter_parts.append(cel.in_("user_defined_function_id", ids)) + query_filter = cel.and_(*filter_parts) + + functions = await self._low_level_client.list_all_functions( + query_filter=query_filter or None, + order_by=order_by, + max_results=limit, + **({"page_size": page_size} if page_size is not None else {}), + ) + return self._apply_client_to_instances(functions) + + async def find(self, **kwargs) -> UserDefinedFunction | None: + """Find one function. Takes the same arguments as `list_`. + + Raises if more than one matches. + + Args: + **kwargs: Keyword arguments to pass to `list_`. + + Returns: + The UserDefinedFunction found or None. + """ + functions = await self.list_(**kwargs) + if len(functions) > 1: + raise ValueError(f"Multiple ({len(functions)}) functions found for query") + elif len(functions) == 1: + return functions[0] + return None + + async def create(self, create: UserDefinedFunctionCreate | dict) -> UserDefinedFunction: + """Create a new function. + + Args: + create: The function definition. + + Returns: + The created UserDefinedFunction. + """ + if isinstance(create, dict): + create = UserDefinedFunctionCreate.model_validate(create) + created = await self._low_level_client.create_function(create=create) + return self._apply_client_to_instance(created) + + async def update( + self, + user_defined_function: str | UserDefinedFunction, + update: UserDefinedFunctionUpdate | dict, + change_notes: str | None = None, + ) -> UserDefinedFunction: + """Update a function. + + Changes to the expression, inputs, description, or metadata create a new version. + + Args: + user_defined_function: The UserDefinedFunction or function ID to update. + update: Updates to apply to the function. + change_notes: A note to attach to the new version. The server treats an empty + note as a change, so the current version's note carries over when omitted. + + Returns: + The updated UserDefinedFunction. + """ + current = ( + user_defined_function + if isinstance(user_defined_function, UserDefinedFunction) + else await self.get(user_defined_function_id=user_defined_function) + ) + if isinstance(update, dict): + update = UserDefinedFunctionUpdate.model_validate(update) + update.resource_id = current._id_or_error + updated = await self._low_level_client.update_function( + update, change_notes=change_notes or current.latest_version.change_notes + ) + return self._apply_client_to_instance(updated) + + async def validate_expression( + self, expression: str, function_inputs: list[FunctionInput] + ) -> UserDefinedFunctionValidation: + """Check an expression without saving it. + + Args: + expression: The expression to check. + function_inputs: The inputs the expression refers to. The server rejects an + empty list. + + Returns: + Whether the expression compiles, and its output type or error. + """ + return await self._low_level_client.validate_function( + expression=expression, function_inputs=function_inputs + ) + + async def get_where_used( + self, + user_defined_function: str | UserDefinedFunction | None = None, + *, + version: str | UserDefinedFunctionVersion | None = None, + ) -> FunctionUsage: + """Get what uses a function. + + Check this before changing inputs or the output type. The server refuses those + changes once a function is in use. + + Each result set is capped at 1000. The service pages them separately but accepts + no page token, so the rest cannot be fetched; `FunctionUsage.truncated` says when + that happened. + + Args: + user_defined_function: The UserDefinedFunction or function ID. + version: A specific UserDefinedFunctionVersion or version ID, instead of the + function as a whole. + + Returns: + The functions, calculated channels, and rules that use it. + + Raises: + ValueError: If neither or both are provided. + """ + if (user_defined_function is None) == (version is None): + raise ValueError("Exactly one of user_defined_function or version must be provided") + if version is not None: + version_id = ( + version._id_or_error if isinstance(version, UserDefinedFunctionVersion) else version + ) + return self._apply_client_to_usage( + await self._low_level_client.get_dependents(version_id=version_id) + ) + function_id = ( + user_defined_function._id_or_error + if isinstance(user_defined_function, UserDefinedFunction) + else user_defined_function + ) + return self._apply_client_to_usage( + await self._low_level_client.get_dependents(function_id=function_id) + ) + + def _apply_client_to_usage(self, usage: FunctionUsage) -> FunctionUsage: + """Hand the client to every object in the usage, so their properties resolve.""" + self._apply_client_to_instances(usage.functions) + self._apply_client_to_instances(usage.calculated_channels) + self._apply_client_to_instances(usage.rules) + return usage + + async def archive( + self, user_defined_function: str | UserDefinedFunction + ) -> UserDefinedFunction: + """Archive a function. + + Args: + user_defined_function: The UserDefinedFunction or function ID to archive. + + Returns: + The archived UserDefinedFunction. + """ + return await self.update(user_defined_function, UserDefinedFunctionUpdate(is_archived=True)) + + async def unarchive( + self, user_defined_function: str | UserDefinedFunction + ) -> UserDefinedFunction: + """Unarchive a function. + + Args: + user_defined_function: The UserDefinedFunction or function ID to unarchive. + + Returns: + The unarchived UserDefinedFunction. + """ + return await self.update( + user_defined_function, UserDefinedFunctionUpdate(is_archived=False) + ) diff --git a/python/lib/sift_client/sift_types/__init__.py b/python/lib/sift_client/sift_types/__init__.py index f844cbc25f..3f806c3903 100644 --- a/python/lib/sift_client/sift_types/__init__.py +++ b/python/lib/sift_client/sift_types/__init__.py @@ -230,6 +230,16 @@ TestStepType, ) from sift_client.sift_types.user import User, UserOrganization +from sift_client.sift_types.user_defined_function import ( + FunctionDataType, + FunctionInput, + FunctionUsage, + UserDefinedFunction, + UserDefinedFunctionCreate, + UserDefinedFunctionUpdate, + UserDefinedFunctionValidation, + UserDefinedFunctionVersion, +) if "pytest" in sys.modules: # These are not test classes, so we need to set __test__ to False to avoid pytest warnings. @@ -276,6 +286,9 @@ "DataImportStatusDetails", "Flow", "FlowConfig", + "FunctionDataType", + "FunctionInput", + "FunctionUsage", "IngestionConfig", "IngestionConfigCreate", "Job", @@ -334,5 +347,10 @@ "TestStepCreate", "TestStepType", "User", + "UserDefinedFunction", + "UserDefinedFunctionCreate", + "UserDefinedFunctionUpdate", + "UserDefinedFunctionValidation", + "UserDefinedFunctionVersion", "UserOrganization", ] diff --git a/python/lib/sift_client/sift_types/user_defined_function.py b/python/lib/sift_client/sift_types/user_defined_function.py new file mode 100644 index 0000000000..82be346caf --- /dev/null +++ b/python/lib/sift_client/sift_types/user_defined_function.py @@ -0,0 +1,328 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from enum import Enum +from typing import TYPE_CHECKING, ClassVar + +from pydantic import BaseModel +from sift.common.type.v1.user_defined_functions_pb2 import ( + FunctionDataType as FunctionDataTypeProto, +) +from sift.common.type.v1.user_defined_functions_pb2 import ( + FunctionInput as FunctionInputProto, +) +from sift.common.type.v1.user_defined_functions_pb2 import ( + UserDefinedFunction as UserDefinedFunctionProto, +) +from sift.user_defined_functions.v1.user_defined_functions_pb2 import ( + CreateUserDefinedFunctionRequest as CreateUserDefinedFunctionRequestProto, +) + +from sift_client.sift_types._base import ( + BaseType, + MappingHelper, + ModelCreate, + ModelCreateUpdateBase, + ModelUpdate, +) +from sift_client.sift_types.calculated_channel import CalculatedChannel # noqa: TC001 +from sift_client.sift_types.rule import Rule # noqa: TC001 +from sift_client.util.metadata import metadata_dict_to_proto, metadata_proto_to_dict + +if TYPE_CHECKING: + from sift_client.client import SiftClient + from sift_client.sift_types.user import User + + +class FunctionDataType(Enum): + """Enum for the data type of a function input or output.""" + + UNSPECIFIED = FunctionDataTypeProto.FUNCTION_DATA_TYPE_UNSPECIFIED # 0 + NUMERIC = FunctionDataTypeProto.FUNCTION_DATA_TYPE_NUMERIC # 1 + STRING = FunctionDataTypeProto.FUNCTION_DATA_TYPE_STRING # 2 + BOOL = FunctionDataTypeProto.FUNCTION_DATA_TYPE_BOOL # 3 + + +class FunctionInput(BaseModel): + """An input the function's expression refers to. + + Attributes: + identifier: How the expression refers to this input, such as `$1`. + data_type: The type the input accepts. + scalar: Whether the input is a literal value rather than a channel. + """ + + identifier: str + data_type: FunctionDataType = FunctionDataType.NUMERIC + scalar: bool = False + + @classmethod + def _from_proto(cls, proto: FunctionInputProto) -> FunctionInput: + return cls( + identifier=proto.identifier, + data_type=FunctionDataType(proto.data_type), + scalar=proto.constant, + ) + + +def _function_input_to_proto(**kwargs) -> FunctionInputProto: + """Convert a function input dict (from model_dump) into its proto form.""" + data_type = kwargs.get("data_type", FunctionDataType.NUMERIC) + return FunctionInputProto( + identifier=kwargs["identifier"], + data_type=data_type.value if isinstance(data_type, FunctionDataType) else data_type, + constant=kwargs.get("scalar", False), + ) + + +class UserDefinedFunctionVersion(BaseType[UserDefinedFunctionProto, "UserDefinedFunctionVersion"]): + """One saved revision of a function. `id_` is the version ID.""" + + user_defined_function_id: str + version: int + expression: str + change_message: str + change_notes: str + function_inputs: list[FunctionInput] + dependency_version_ids: list[str] + created_date: datetime + created_by_user_id: str + output_type: FunctionDataType | None + + @classmethod + def _from_proto( + cls, proto: UserDefinedFunctionProto, sift_client: SiftClient | None = None + ) -> UserDefinedFunctionVersion: + return cls( + proto=proto, + id_=proto.user_defined_function_version_id, + user_defined_function_id=proto.user_defined_function_id, + version=proto.version, + expression=proto.expression, + change_message=proto.change_message, + change_notes=proto.user_notes, + function_inputs=[FunctionInput._from_proto(i) for i in proto.function_inputs], + dependency_version_ids=[ + d.user_defined_function_version_id for d in proto.function_dependencies + ], + created_date=proto.created_date.ToDatetime(tzinfo=timezone.utc), + created_by_user_id=proto.created_by_user_id, + output_type=( + FunctionDataType(proto.function_output_type) if proto.function_output_type else None + ), + _client=sift_client, + ) + + @property + def function(self) -> UserDefinedFunction: + """Fetch the function this version belongs to.""" + return self.client.user_defined_functions.get( + user_defined_function_id=self.user_defined_function_id + ) + + @property + def created_by(self) -> User: + """Fetch the User that created this version.""" + return self.client.users.get(user_id=self.created_by_user_id) + + @property + def dependencies(self) -> list[UserDefinedFunctionVersion]: + """Fetch the function versions this one calls.""" + if not self.dependency_version_ids: + return [] + return self.client.user_defined_functions.versions.batch_get( + versions=self.dependency_version_ids + ) + + +class UserDefinedFunction(BaseType[UserDefinedFunctionProto, "UserDefinedFunction"]): + """A reusable expression that calculated channels and rules can call. + + Version agnostic. The expression, inputs, and output type live on + `latest_version`; each save produces a new one. + """ + + name: str + description: str + metadata: dict[str, str | float | bool] + created_date: datetime + modified_date: datetime + created_by_user_id: str + modified_by_user_id: str + is_archived: bool + latest_version: UserDefinedFunctionVersion + + archived_date: datetime | None + + @classmethod + def _from_proto( + cls, proto: UserDefinedFunctionProto, sift_client: SiftClient | None = None + ) -> UserDefinedFunction: + return cls( + proto=proto, + id_=proto.user_defined_function_id, + name=proto.name, + description=proto.description, + metadata=metadata_proto_to_dict(proto.metadata), # type: ignore + created_date=proto.created_date.ToDatetime(tzinfo=timezone.utc), + modified_date=proto.modified_date.ToDatetime(tzinfo=timezone.utc), + created_by_user_id=proto.created_by_user_id, + modified_by_user_id=proto.modified_by_user_id, + is_archived=proto.is_archived, + latest_version=UserDefinedFunctionVersion._from_proto(proto, sift_client), + archived_date=( + proto.archived_date.ToDatetime(tzinfo=timezone.utc) + if proto.HasField("archived_date") + else None + ), + _client=sift_client, + ) + + @property + def versions(self) -> list[UserDefinedFunctionVersion]: + """Fetch every version of this function, newest first.""" + return self.client.user_defined_functions.versions.list_( + user_defined_function=self, order_by="version desc" + ) + + @property + def created_by(self) -> User: + """Fetch the User that created this function.""" + return self.client.users.get(user_id=self.created_by_user_id) + + @property + def modified_by(self) -> User: + """Fetch the User that last modified this function.""" + return self.client.users.get(user_id=self.modified_by_user_id) + + def update(self, update: UserDefinedFunctionUpdate | dict) -> UserDefinedFunction: + """Update the function. This creates a new version. + + Args: + update: The update to apply. See UserDefinedFunctionUpdate for updatable fields. + + Returns: + The updated function. + """ + updated = self.client.user_defined_functions.update( + user_defined_function=self, update=update + ) + self._update(updated) + return self + + def archive(self) -> UserDefinedFunction: + """Archive the function.""" + updated = self.client.user_defined_functions.archive(user_defined_function=self) + self._update(updated) + return self + + def unarchive(self) -> UserDefinedFunction: + """Unarchive the function.""" + updated = self.client.user_defined_functions.unarchive(user_defined_function=self) + self._update(updated) + return self + + +class UserDefinedFunctionBase(ModelCreateUpdateBase): + """Base class for UserDefinedFunction create and update models.""" + + description: str | None = None + metadata: dict[str, str | float | bool] | None = None + + _to_proto_helpers: ClassVar[dict[str, MappingHelper]] = { + "metadata": MappingHelper( + proto_attr_path="metadata", + update_field="metadata", + converter=metadata_dict_to_proto, + ), + } + + +class UserDefinedFunctionCreate( + UserDefinedFunctionBase, ModelCreate[CreateUserDefinedFunctionRequestProto] +): + """Create model for UserDefinedFunction.""" + + name: str + expression: str + """The CEL expression, referring to inputs positionally as `$1`, `$2`.""" + function_inputs: list[FunctionInput] + change_notes: str | None = None + + _to_proto_helpers: ClassVar[dict[str, MappingHelper]] = { + **UserDefinedFunctionBase._to_proto_helpers, + "function_inputs": MappingHelper( + proto_attr_path="function_inputs", + update_field="function_inputs", + converter=_function_input_to_proto, # type: ignore[arg-type] + ), + "change_notes": MappingHelper(proto_attr_path="user_notes", update_field="user_notes"), + } + + def _get_proto_class(self) -> type[CreateUserDefinedFunctionRequestProto]: + return CreateUserDefinedFunctionRequestProto + + +class UserDefinedFunctionUpdate(UserDefinedFunctionBase, ModelUpdate[UserDefinedFunctionProto]): + """Update model for UserDefinedFunction. + + Once a function has dependents the server refuses to change `name`, + `function_inputs`, or the output type. Check `get_where_used` first. + """ + + name: str | None = None + expression: str | None = None + function_inputs: list[FunctionInput] | None = None + is_archived: bool | None = None + + _to_proto_helpers: ClassVar[dict[str, MappingHelper]] = { + **UserDefinedFunctionBase._to_proto_helpers, + "function_inputs": MappingHelper( + proto_attr_path="function_inputs", + update_field="function_inputs", + converter=_function_input_to_proto, # type: ignore[arg-type] + ), + } + + def _get_proto_class(self) -> type[UserDefinedFunctionProto]: + return UserDefinedFunctionProto + + def _add_resource_id_to_proto(self, proto_msg: UserDefinedFunctionProto): + if self._resource_id is None: + raise ValueError("Resource ID must be set before adding to proto") + proto_msg.user_defined_function_id = self._resource_id + + +class UserDefinedFunctionValidation(BaseModel): + """The result of checking an expression before saving it. + + Attributes: + is_valid: Whether the expression compiles. + error: Why it failed, when it is not valid. + output_type: The type the expression returns, when it is valid. + """ + + is_valid: bool + error: str | None = None + output_type: FunctionDataType | None = None + + +class FunctionUsage(BaseModel): + """Where a function is used. + + Attributes: + functions: Other functions that call it. + calculated_channels: Calculated channels that use it. + rules: Rules that use it. + truncated: Whether the service had more to return than it sent. + """ + + functions: list[UserDefinedFunction] = [] + calculated_channels: list[CalculatedChannel] = [] + rules: list[Rule] = [] + truncated: bool = False + + @property + def is_used(self) -> bool: + """Whether anything uses the function.""" + return bool(self.functions or self.calculated_channels or self.rules) diff --git a/python/lib/sift_client/util/util.py b/python/lib/sift_client/util/util.py index 73253bcd3b..a55d8b4e12 100644 --- a/python/lib/sift_client/util/util.py +++ b/python/lib/sift_client/util/util.py @@ -19,6 +19,7 @@ RunsAPIAsync, TagsAPIAsync, TestResultsAPIAsync, + UserDefinedFunctionsAPIAsync, UsersAPIAsync, ) from sift_client.resources.access_control import AccessControlAPIAsync @@ -69,6 +70,9 @@ class AsyncAPIs(NamedTuple): test_results: TestResultsAPIAsync """Instance of the Test Results API for making asynchronous requests.""" + user_defined_functions: UserDefinedFunctionsAPIAsync + """Instance of the User Defined Functions API for making asynchronous requests.""" + users: UsersAPIAsync """Instance of the Users API for making asynchronous requests."""