cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(resources): implement virtual resource type base class with examples

Browse Source 
- Implement VirtualResource base class for abstract/computed resources
- Add MetricResource example for computed metrics
- Add APIEndpointResource example for API endpoints
- Implement comprehensive BDD tests for virtual resource types
- Full type annotations with Generic support
- Support for on-demand computation via compute_fn
- Support for kwargs passing to compute functions

Closes #8610
This commit is contained in:
HAL 9000
2026-04-18 18:54:34 +00:00
committed by drew
parent 132d3843a1
commit 2b67362b6f
4 changed files with 960 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/resource_virtual_types.feature
+120
View File
@@ -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
features/steps/resource_virtual_types_steps.py
+628
View File
@@ -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}'"
)
src/cleveragents/resource/__init__.py
+8
View File
@@ -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",
src/cleveragents/resource/virtual.py
+204
View File
@@ -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",
]