diff --git a/features/resource_virtual_types.feature b/features/resource_virtual_types.feature new file mode 100644 index 000000000..3df1ab54c --- /dev/null +++ b/features/resource_virtual_types.feature @@ -0,0 +1,120 @@ +Feature: Virtual Resource Type Base Class and Implementations + As a developer + I want to use virtual resources to represent computed/abstract resources + So that I can work with non-physical resources like metrics and API endpoints + + Background: + Given I have imported the virtual resource module + + Scenario: Create a basic VirtualResource with valid parameters + When I create a VirtualResource with: + | name | test-metric | + | description | A test metric resource | + | compute_fn | lambda: 42.0 | + Then the VirtualResource should have: + | name | test-metric | + | description | A test metric resource | + And the VirtualResource should be callable + + Scenario: Compute a virtual resource value on demand + Given I have a VirtualResource named "cpu-usage" with compute_fn returning 45.2 + When I compute the virtual resource + Then the computed value should be 45.2 + + Scenario: Reject invalid resource names + When I try to create a VirtualResource with invalid name "123-invalid" + Then it should raise ValueError with message containing "invalid characters" + + Scenario: Reject empty resource names + When I try to create a VirtualResource with empty name "" + Then it should raise ValueError with message containing "non-empty string" + + Scenario: Reject non-callable compute functions + When I try to create a VirtualResource with non-callable compute_fn "not-a-function" + Then it should raise TypeError with message containing "callable" + + Scenario: Create a MetricResource with unit + When I create a MetricResource with: + | name | memory-usage | + | description | Current memory usage | + | compute_fn | lambda: 78.5 | + | unit | percent | + Then the MetricResource should have: + | name | memory-usage | + | description | Current memory usage | + | unit | percent | + + Scenario: Compute a MetricResource value + Given I have a MetricResource named "disk-usage" with compute_fn returning 62.3 and unit "percent" + When I compute the metric resource + Then the computed value should be 62.3 + + Scenario: Create an APIEndpointResource with URL + When I create an APIEndpointResource with: + | name | api-data | + | description | Data from external API | + | compute_fn | lambda: {"status": "ok"} | + | endpoint_url | https://api.example.com/data | + Then the APIEndpointResource should have: + | name | api-data | + | description | Data from external API | + | endpoint_url | https://api.example.com/data | + + Scenario: Compute an APIEndpointResource value + Given I have an APIEndpointResource named "api-status" with compute_fn returning {"status": "ok", "code": 200} + When I compute the API endpoint resource + Then the API endpoint computed value is a dict with keys "status,code" + And the API endpoint value["status"] is "ok" + And the API endpoint value["code"] is 200 + + Scenario: Pass kwargs to compute function + Given I have a VirtualResource with compute_fn that accepts kwargs + When I compute the virtual resource with kwargs {"multiplier": 2} + Then the computed value should reflect the passed kwargs + + Scenario: Virtual resource string representation + Given I have a VirtualResource named "test-resource" with description "Test resource" + When I get the string representation of the virtual resource + Then it should contain "VirtualResource" + And it should contain "test-resource" + And it should contain "Test resource" + + Scenario: MetricResource string representation + Given I have a MetricResource named "cpu-usage" with unit "percent" + When I get the string representation of the virtual resource + Then it should contain "MetricResource" + And it should contain "cpu-usage" + And it should contain "percent" + + Scenario: APIEndpointResource string representation + Given I have an APIEndpointResource named "api-data" with endpoint_url "https://api.example.com" + When I get the string representation of the virtual resource + Then it should contain "APIEndpointResource" + And it should contain "api-data" + And it should contain "https://api.example.com" + + Scenario: Virtual resource with complex compute function + Given I have a VirtualResource with a complex compute function + When I compute the virtual resource + Then the complex computed value should be correct + + Scenario: Multiple virtual resources can coexist + Given I have created 3 different virtual resources + When I compute all of them + Then each should return its expected value + And they should not interfere with each other + + Scenario: Virtual resource compute function exception handling + Given I have a VirtualResource with compute_fn that raises an exception + When I compute the virtual resource + Then it should raise the exception from compute_fn + + Scenario: Virtual resource with default kwargs + Given I have a VirtualResource with compute_fn that has default kwargs + When I compute without providing kwargs + Then it should use the default values + + Scenario: Virtual resource inheritance + Given I have created a custom VirtualResource subclass + When I instantiate and compute it + Then it should work correctly with the custom implementation diff --git a/features/steps/resource_virtual_types_steps.py b/features/steps/resource_virtual_types_steps.py new file mode 100644 index 000000000..0fcbdacdf --- /dev/null +++ b/features/steps/resource_virtual_types_steps.py @@ -0,0 +1,628 @@ +"""Step definitions for resource_virtual_types.feature. + +Tests the VirtualResource base class and example implementations +(MetricResource, APIEndpointResource). +""" + +from __future__ import annotations + +from typing import Any + +from behave import given, then, when + +from cleveragents.resource.virtual import ( + APIEndpointResource, + MetricResource, + VirtualResource, +) + + +# ──────────────────────────────────────────────────────────── +# Background steps +# ──────────────────────────────────────────────────────────── + + +@given("I have imported the virtual resource module") +def step_import_virtual_module(context: object) -> None: + """Verify the virtual resource module is imported.""" + assert VirtualResource is not None + assert MetricResource is not None + assert APIEndpointResource is not None + + +# ──────────────────────────────────────────────────────────── +# VirtualResource creation steps +# ──────────────────────────────────────────────────────────── + + +@when("I create a VirtualResource with:") +def step_create_virtual_resource(context: object) -> None: + """Create a VirtualResource from table data.""" + table_dict = {row["name"]: row["value"] for row in context.table} + + name = table_dict.get("name", "test-resource") + description = table_dict.get("description", "Test resource") + compute_fn_str = table_dict.get("compute_fn", "lambda: 42") + + compute_fn = eval(compute_fn_str) # noqa: S307 + + resource = VirtualResource( + name=name, + description=description, + compute_fn=compute_fn, + ) + + context.virtual_resource = resource + + +@then("the VirtualResource should have:") +def step_check_virtual_resource_attrs(context: object) -> None: + """Check VirtualResource attributes.""" + resource = context.virtual_resource + + for row in context.table: + attr_name = row["name"] + expected_value = row["value"] + + actual_value = getattr(resource, attr_name) + assert actual_value == expected_value, ( + f"Expected {attr_name}={expected_value!r}, got {actual_value!r}" + ) + + +@then("the VirtualResource should be callable") +def step_virtual_resource_callable(context: object) -> None: + """Check that VirtualResource has a compute method.""" + resource = context.virtual_resource + assert hasattr(resource, "compute"), "VirtualResource should have compute method" + assert callable(resource.compute), "compute should be callable" + + +# ──────────────────────────────────────────────────────────── +# VirtualResource computation steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource named {name!r} with compute_fn returning {value}") +def step_create_virtual_with_value(context: object, name: str, value: str) -> None: + """Create a VirtualResource that returns a specific value.""" + if value.lower() == "true": + computed_value = True + elif value.lower() == "false": + computed_value = False + elif value.isdigit(): + computed_value = int(value) + else: + try: + computed_value = float(value) + except ValueError: + computed_value = value + + resource = VirtualResource( + name=name, + description=f"Test resource {name}", + compute_fn=lambda: computed_value, + ) + context.virtual_resource = resource + + +@when("I compute the virtual resource") +def step_compute_virtual_resource(context: object) -> None: + """Compute the virtual resource value.""" + resource = context.virtual_resource + context.computed_value = resource.compute() + + +@then("the computed value should be {expected}") +def step_check_computed_value(context: object, expected: str) -> None: + """Check the computed value.""" + if expected.lower() == "true": + expected_value = True + elif expected.lower() == "false": + expected_value = False + elif expected.isdigit(): + expected_value = int(expected) + else: + try: + expected_value = float(expected) + except ValueError: + expected_value = expected + + actual = context.computed_value + assert actual == expected_value, ( + f"Expected {expected_value!r}, got {actual!r}" + ) + + +# ──────────────────────────────────────────────────────────── +# Error handling steps +# ──────────────────────────────────────────────────────────── + + +@when("I try to create a VirtualResource with invalid name {name!r}") +def step_create_with_invalid_name(context: object, name: str) -> None: + """Try to create a VirtualResource with invalid name.""" + try: + VirtualResource( + name=name, + description="Test", + compute_fn=lambda: 42, + ) + context.creation_error = None + except (ValueError, TypeError) as exc: + context.creation_error = exc + + +@when("I try to create a VirtualResource with empty name {name!r}") +def step_create_with_empty_name(context: object, name: str) -> None: + """Try to create a VirtualResource with empty name.""" + try: + VirtualResource( + name=name, + description="Test", + compute_fn=lambda: 42, + ) + context.creation_error = None + except (ValueError, TypeError) as exc: + context.creation_error = exc + + +@when("I try to create a VirtualResource with non-callable compute_fn {fn!r}") +def step_create_with_non_callable(context: object, fn: str) -> None: + """Try to create a VirtualResource with non-callable compute_fn.""" + try: + VirtualResource( + name="test", + description="Test", + compute_fn=fn, # type: ignore + ) + context.creation_error = None + except (ValueError, TypeError) as exc: + context.creation_error = exc + + +@then("it should raise {error_type} with message containing {message!r}") +def step_check_error(context: object, error_type: str, message: str) -> None: + """Check that the expected error was raised.""" + error = context.creation_error + assert error is not None, f"Expected {error_type} but no error was raised" + + error_class_name = error.__class__.__name__ + assert error_class_name == error_type, ( + f"Expected {error_type}, got {error_class_name}" + ) + + assert message.lower() in str(error).lower(), ( + f"Error message does not contain '{message}': {error}" + ) + + +# ──────────────────────────────────────────────────────────── +# MetricResource steps +# ──────────────────────────────────────────────────────────── + + +@when("I create a MetricResource with:") +def step_create_metric_resource(context: object) -> None: + """Create a MetricResource from table data.""" + table_dict = {row["name"]: row["value"] for row in context.table} + + name = table_dict.get("name", "test-metric") + description = table_dict.get("description", "Test metric") + compute_fn_str = table_dict.get("compute_fn", "lambda: 42.0") + unit = table_dict.get("unit", "") + + compute_fn = eval(compute_fn_str) # noqa: S307 + + resource = MetricResource( + name=name, + description=description, + compute_fn=compute_fn, + unit=unit, + ) + + context.metric_resource = resource + + +@then("the MetricResource should have:") +def step_check_metric_attrs(context: object) -> None: + """Check MetricResource attributes.""" + resource = context.metric_resource + + for row in context.table: + attr_name = row["name"] + expected_value = row["value"] + + actual_value = getattr(resource, attr_name) + assert actual_value == expected_value, ( + f"Expected {attr_name}={expected_value!r}, got {actual_value!r}" + ) + + +@given("I have a MetricResource named {name!r} with compute_fn returning {value} and unit {unit!r}") +def step_create_metric_with_value(context: object, name: str, value: str, unit: str) -> None: + """Create a MetricResource that returns a specific value.""" + computed_value = float(value) + + resource = MetricResource( + name=name, + description=f"Test metric {name}", + compute_fn=lambda: computed_value, + unit=unit, + ) + context.metric_resource = resource + + +@when("I compute the metric resource") +def step_compute_metric_resource(context: object) -> None: + """Compute the metric resource value.""" + resource = context.metric_resource + context.computed_value = resource.compute() + + +# ──────────────────────────────────────────────────────────── +# APIEndpointResource steps +# ──────────────────────────────────────────────────────────── + + +@when("I create an APIEndpointResource with:") +def step_create_api_endpoint_resource(context: object) -> None: + """Create an APIEndpointResource from table data.""" + table_dict = {row["name"]: row["value"] for row in context.table} + + name = table_dict.get("name", "test-api") + description = table_dict.get("description", "Test API") + compute_fn_str = table_dict.get("compute_fn", "lambda: {}") + endpoint_url = table_dict.get("endpoint_url", "") + + compute_fn = eval(compute_fn_str) # noqa: S307 + + resource = APIEndpointResource( + name=name, + description=description, + compute_fn=compute_fn, + endpoint_url=endpoint_url, + ) + + context.api_resource = resource + + +@then("the APIEndpointResource should have:") +def step_check_api_attrs(context: object) -> None: + """Check APIEndpointResource attributes.""" + resource = context.api_resource + + for row in context.table: + attr_name = row["name"] + expected_value = row["value"] + + actual_value = getattr(resource, attr_name) + assert actual_value == expected_value, ( + f"Expected {attr_name}={expected_value!r}, got {actual_value!r}" + ) + + +@given("I have an APIEndpointResource named {name!r} with compute_fn returning {value}") +def step_create_api_with_value(context: object, name: str, value: str) -> None: + """Create an APIEndpointResource that returns a specific value.""" + computed_value = eval(value) # noqa: S307 + + resource = APIEndpointResource( + name=name, + description=f"Test API {name}", + compute_fn=lambda: computed_value, + ) + context.api_resource = resource + + +@when("I compute the API endpoint resource") +def step_compute_api_resource(context: object) -> None: + """Compute the API endpoint resource value.""" + resource = context.api_resource + context.computed_value = resource.compute() + + +@then("the API endpoint computed value is a dict with keys {keys!r}") +def step_api_check_dict_keys(context: object, keys: str) -> None: + """Check that computed value is a dict with expected keys.""" + value = context.computed_value + assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}" + + expected_keys = [k.strip() for k in keys.strip('"').split(",")] + + for key in expected_keys: + assert key in value, f"Expected key '{key}' not in dict: {value.keys()}" + + +@then("the API endpoint value[{key!r}] is {expected!r}") +def step_api_check_dict_value(context: object, key: str, expected: str) -> None: + """Check a specific value in the computed dict.""" + value = context.computed_value + assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}" + + if expected.lower() == "true": + expected_value = True + elif expected.lower() == "false": + expected_value = False + elif expected.isdigit(): + expected_value = int(expected) + else: + try: + expected_value = float(expected) + except ValueError: + expected_value = expected + + actual = value.get(key) + assert actual == expected_value, ( + f"Expected {key}={expected_value!r}, got {actual!r}" + ) + + +# ──────────────────────────────────────────────────────────── +# Kwargs handling steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource with compute_fn that accepts kwargs") +def step_create_virtual_with_kwargs(context: object) -> None: + """Create a VirtualResource with a compute_fn that accepts kwargs.""" + def compute_with_kwargs(multiplier: int = 1) -> int: + return 42 * multiplier + + resource = VirtualResource( + name="test-kwargs", + description="Test resource with kwargs", + compute_fn=compute_with_kwargs, + ) + context.virtual_resource = resource + + +@when("I compute the virtual resource with kwargs {kwargs_str}") +def step_compute_with_kwargs(context: object, kwargs_str: str) -> None: + """Compute the virtual resource with kwargs.""" + kwargs = eval(kwargs_str) # noqa: S307 + resource = context.virtual_resource + context.computed_value = resource.compute(**kwargs) + + +@then("the computed value should reflect the passed kwargs") +def step_check_kwargs_reflected(context: object) -> None: + """Check that kwargs were properly used.""" + assert context.computed_value == 84, ( + f"Expected 84, got {context.computed_value}" + ) + + +# ──────────────────────────────────────────────────────────── +# String representation steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource named {name!r} with description {description!r}") +def step_create_virtual_for_repr(context: object, name: str, description: str) -> None: + """Create a VirtualResource for string representation testing.""" + resource = VirtualResource( + name=name, + description=description, + compute_fn=lambda: 42, + ) + context.virtual_resource = resource + + +@when("I get the string representation of the virtual resource") +def step_get_string_repr(context: object) -> None: + """Get the string representation of the resource.""" + resource = context.virtual_resource or context.metric_resource or context.api_resource + context.string_repr = repr(resource) + + +@then("it should contain {text!r}") +def step_check_repr_contains(context: object, text: str) -> None: + """Check that string representation contains text.""" + assert text in context.string_repr, ( + f"Expected '{text}' in repr: {context.string_repr}" + ) + + +@given("I have a MetricResource named {name!r} with unit {unit!r}") +def step_create_metric_for_repr(context: object, name: str, unit: str) -> None: + """Create a MetricResource for string representation testing.""" + resource = MetricResource( + name=name, + description="Test metric", + compute_fn=lambda: 42.0, + unit=unit, + ) + context.metric_resource = resource + + +@given("I have an APIEndpointResource named {name!r} with endpoint_url {url!r}") +def step_create_api_for_repr(context: object, name: str, url: str) -> None: + """Create an APIEndpointResource for string representation testing.""" + resource = APIEndpointResource( + name=name, + description="Test API", + compute_fn=lambda: {}, + endpoint_url=url, + ) + context.api_resource = resource + + +# ──────────────────────────────────────────────────────────── +# Complex compute function steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource with a complex compute function") +def step_create_complex_virtual(context: object) -> None: + """Create a VirtualResource with a complex compute function.""" + def complex_compute() -> dict[str, Any]: + return { + "result": 42, + "nested": {"value": "test"}, + "list": [1, 2, 3], + } + + resource = VirtualResource( + name="complex-resource", + description="Complex resource", + compute_fn=complex_compute, + ) + context.virtual_resource = resource + + +@then("the computed value should be correct") +def step_check_complex_value(context: object) -> None: + """Check that complex computed value is correct.""" + value = context.computed_value + assert isinstance(value, dict), f"Expected dict, got {type(value).__name__}" + assert value["result"] == 42 + assert value["nested"]["value"] == "test" + assert value["list"] == [1, 2, 3] + + +# ──────────────────────────────────────────────────────────── +# Multiple resources steps +# ──────────────────────────────────────────────────────────── + + +@given("I have created {count:d} different virtual resources") +def step_create_multiple_resources(context: object, count: int) -> None: + """Create multiple different virtual resources.""" + resources = [] + for i in range(count): + resource = VirtualResource( + name=f"resource-{i}", + description=f"Resource {i}", + compute_fn=lambda i=i: i * 10, + ) + resources.append(resource) + + context.resources = resources + + +@when("I compute all of them") +def step_compute_all_resources(context: object) -> None: + """Compute all resources.""" + context.computed_values = [r.compute() for r in context.resources] + + +@then("each should return its expected value") +def step_check_each_value(context: object) -> None: + """Check that each resource returned its expected value.""" + for i, value in enumerate(context.computed_values): + expected = i * 10 + assert value == expected, ( + f"Resource {i}: expected {expected}, got {value}" + ) + + +@then("they should not interfere with each other") +def step_check_no_interference(context: object) -> None: + """Check that resources don't interfere with each other.""" + assert len(context.computed_values) == len(context.resources) + + +# ──────────────────────────────────────────────────────────── +# Exception handling steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource with compute_fn that raises an exception") +def step_create_virtual_with_exception(context: object) -> None: + """Create a VirtualResource with compute_fn that raises an exception.""" + def compute_with_error() -> None: + raise RuntimeError("Test error") + + resource = VirtualResource( + name="error-resource", + description="Resource that raises", + compute_fn=compute_with_error, + ) + context.virtual_resource = resource + + +@then("it should raise the exception from compute_fn") +def step_check_exception_raised(context: object) -> None: + """Check that the exception was raised.""" + try: + context.virtual_resource.compute() + assert False, "Expected RuntimeError but no exception was raised" + except RuntimeError as exc: + assert "Test error" in str(exc) + + +# ──────────────────────────────────────────────────────────── +# Default kwargs steps +# ──────────────────────────────────────────────────────────── + + +@given("I have a VirtualResource with compute_fn that has default kwargs") +def step_create_virtual_with_defaults(context: object) -> None: + """Create a VirtualResource with default kwargs.""" + def compute_with_defaults(value: int = 100) -> int: + return value + + resource = VirtualResource( + name="default-resource", + description="Resource with defaults", + compute_fn=compute_with_defaults, + ) + context.virtual_resource = resource + + +@when("I compute without providing kwargs") +def step_compute_without_kwargs(context: object) -> None: + """Compute without providing kwargs.""" + context.computed_value = context.virtual_resource.compute() + + +@then("it should use the default values") +def step_check_defaults_used(context: object) -> None: + """Check that default values were used.""" + assert context.computed_value == 100, ( + f"Expected 100 (default), got {context.computed_value}" + ) + + +# ──────────────────────────────────────────────────────────── +# Inheritance steps +# ──────────────────────────────────────────────────────────── + + +@given("I have created a custom VirtualResource subclass") +def step_create_custom_subclass(context: object) -> None: + """Create a custom VirtualResource subclass.""" + class CustomResource(VirtualResource[str]): + """Custom virtual resource subclass.""" + + def __init__(self, name: str, description: str) -> None: + super().__init__( + name=name, + description=description, + compute_fn=self._compute, + ) + + def _compute(self) -> str: + return f"Custom: {self.name}" + + resource = CustomResource( + name="custom-resource", + description="Custom implementation", + ) + context.custom_resource = resource + + +@when("I instantiate and compute it") +def step_compute_custom(context: object) -> None: + """Compute the custom resource.""" + context.computed_value = context.custom_resource.compute() + + +@then("it should work correctly with the custom implementation") +def step_check_custom_works(context: object) -> None: + """Check that custom implementation works.""" + expected = "Custom: custom-resource" + assert context.computed_value == expected, ( + f"Expected '{expected}', got '{context.computed_value}'" + ) diff --git a/src/cleveragents/resource/__init__.py b/src/cleveragents/resource/__init__.py index 24d7aaffc..e016178d1 100644 --- a/src/cleveragents/resource/__init__.py +++ b/src/cleveragents/resource/__init__.py @@ -13,14 +13,22 @@ from cleveragents.resource.inheritance import ( resolve_inheritance_chain, validate_chain, ) +from cleveragents.resource.virtual import ( + APIEndpointResource, + MetricResource, + VirtualResource, +) __all__ = [ + "APIEndpointResource", "MAX_CHAIN_DEPTH", + "MetricResource", "ResourceTypeCircularInheritanceError", "ResourceTypeInheritanceDepthError", "ResourceTypeParentNotFoundError", "ResourceTypeParentRemovalError", "TypeRegistryMap", + "VirtualResource", "find_subtypes", "is_subtype_of", "resolve_fields", diff --git a/src/cleveragents/resource/virtual.py b/src/cleveragents/resource/virtual.py new file mode 100644 index 000000000..063364903 --- /dev/null +++ b/src/cleveragents/resource/virtual.py @@ -0,0 +1,204 @@ +"""Virtual resource type base class and implementations. + +Virtual resources represent abstract or computed resources that don't map to +physical files. Examples include computed metrics, API endpoints, or derived data. + +This module provides: +- :class:`VirtualResource` — base class for all virtual resources +- :class:`MetricResource` — example: computed metrics +- :class:`APIEndpointResource` — example: API endpoints + +Based on: +- docs/specification.md — Virtual Resource Types +- Issue #8610 — Virtual Resource Type Base Class Implementation +""" + +from __future__ import annotations + +import logging +from abc import ABC +from collections.abc import Callable +from typing import Any, Generic, TypeVar + +logger = logging.getLogger(__name__) + +# Type variable for compute function return values +T = TypeVar("T") + + +class VirtualResource(ABC, Generic[T]): + """Base class for virtual (computed/abstract) resources. + + Virtual resources are abstract or computed resources that don't map to + physical files. They can be computed on demand via a compute function. + + Attributes: + name: Unique identifier for this virtual resource. + description: Human-readable description of the resource. + compute_fn: Callable that computes the resource value on demand. + """ + + def __init__( + self, + name: str, + description: str, + compute_fn: Callable[..., T], + ) -> None: + """Initialize a virtual resource. + + Args: + name: Unique identifier for this virtual resource. + Must be a valid resource name (alphanumeric, hyphens, underscores). + description: Human-readable description of what this resource computes. + compute_fn: Callable that computes the resource value on demand. + Should accept no required arguments (all args should have defaults). + + Raises: + ValueError: If name is empty or invalid. + TypeError: If compute_fn is not callable. + """ + if not name or not isinstance(name, str): + raise ValueError( + f"Resource name must be a non-empty string, got {type(name).__name__}" + ) + + if not name.replace("-", "").replace("_", "").isalnum(): + raise ValueError( + f"Resource name '{name}' contains invalid characters. " + "Use only alphanumeric characters, hyphens, and underscores." + ) + + if not callable(compute_fn): + raise TypeError( + f"compute_fn must be callable, got {type(compute_fn).__name__}" + ) + + self.name: str = name + self.description: str = description + self.compute_fn: Callable[..., T] = compute_fn + + def compute(self, **kwargs: Any) -> T: + """Compute the resource value on demand. + + Args: + **kwargs: Optional keyword arguments to pass to the compute function. + + Returns: + The computed resource value. + + Raises: + Exception: Any exception raised by the compute function. + """ + logger.debug( + "Computing virtual resource", + resource_name=self.name, + kwargs_keys=list(kwargs.keys()), + ) + return self.compute_fn(**kwargs) + + def __repr__(self) -> str: + """Return string representation of the virtual resource.""" + return ( + f"{self.__class__.__name__}(" + f"name={self.name!r}, " + f"description={self.description!r})" + ) + + +class MetricResource(VirtualResource[float]): + """Virtual resource that computes a metric value. + + Example: + >>> def compute_cpu_usage() -> float: + ... # Compute CPU usage percentage + ... return 45.2 + >>> metric = MetricResource( + ... name="cpu-usage", + ... description="Current CPU usage percentage", + ... compute_fn=compute_cpu_usage, + ... ) + >>> value = metric.compute() + >>> print(f"CPU Usage: {value}%") + CPU Usage: 45.2% + """ + + def __init__( + self, + name: str, + description: str, + compute_fn: Callable[..., float], + unit: str = "", + ) -> None: + """Initialize a metric resource. + + Args: + name: Unique identifier for this metric. + description: Human-readable description of the metric. + compute_fn: Callable that computes the metric value (returns float). + unit: Optional unit of measurement (e.g., "percent", "bytes", "ms"). + """ + super().__init__(name, description, compute_fn) + self.unit: str = unit + + def __repr__(self) -> str: + """Return string representation of the metric resource.""" + unit_str = f", unit={self.unit!r}" if self.unit else "" + return ( + f"MetricResource(" + f"name={self.name!r}, " + f"description={self.description!r}{unit_str})" + ) + + +class APIEndpointResource(VirtualResource[dict[str, Any]]): + """Virtual resource that represents an API endpoint. + + Example: + >>> def fetch_api_data() -> dict[str, Any]: + ... # Fetch data from an API + ... return {"status": "ok", "data": [1, 2, 3]} + >>> endpoint = APIEndpointResource( + ... name="api-data", + ... description="Data from external API", + ... compute_fn=fetch_api_data, + ... endpoint_url="https://api.example.com/data", + ... ) + >>> data = endpoint.compute() + >>> print(data["status"]) + ok + """ + + def __init__( + self, + name: str, + description: str, + compute_fn: Callable[..., dict[str, Any]], + endpoint_url: str = "", + ) -> None: + """Initialize an API endpoint resource. + + Args: + name: Unique identifier for this endpoint. + description: Human-readable description of the endpoint. + compute_fn: Callable that fetches data from the endpoint + (returns dict). + endpoint_url: Optional URL of the API endpoint. + """ + super().__init__(name, description, compute_fn) + self.endpoint_url: str = endpoint_url + + def __repr__(self) -> str: + """Return string representation of the API endpoint resource.""" + url_str = f", endpoint_url={self.endpoint_url!r}" if self.endpoint_url else "" + return ( + f"APIEndpointResource(" + f"name={self.name!r}, " + f"description={self.description!r}{url_str})" + ) + + +__all__ = [ + "APIEndpointResource", + "MetricResource", + "VirtualResource", +]