nat.observability.mixin.resource_conflict_mixin

Attributes

logger

Exceptions

ResourceConflictError

Raised when multiple exporter instances would conflict over the same resource.

Classes

ResourceConflictMixin

Abstract mixin for detecting resource conflicts between exporter instances.

Module Contents

logger

exception ResourceConflictError

Bases: ValueError

Raised when multiple exporter instances would conflict over the same resource.

Initialize self. See help(type(self)) for accurate signature.

class ResourceConflictMixin(*args, **kwargs)

Bases: abc.ABC

Abstract mixin for detecting resource conflicts between exporter instances.

This mixin provides a framework for exporters to detect when multiple instances would conflict over the same resources (files, database tables, API endpoints, etc.). Each concrete implementation defines what constitutes a resource conflict for that exporter type.

The mixin maintains class-level registries using weakrefs for automatic cleanup when instances are garbage collected.

_registries: dict[type, dict[str, weakref.ref]]

abstractmethod _get_resource_identifiers() → dict[str, Any]

Return dict of resource_type -> identifier that this instance will use.

Examples:

Files: {“file_path”: “/logs/app.log”, “cleanup_pattern”: “app_*.log”} Phoenix: {“project_endpoint”: “my_project@http://localhost:6006”} Database: {“table_name”: “events”, “connection”: “postgresql://…”}

Returns:

dict[str, Any]: Dict mapping resource type names to unique identifiers for those resources.

abstractmethod _format_conflict_error(

resource_type: str,

identifier: Any,

existing_instance: Any,

) → str

Format a user-friendly error message for a resource conflict.

Args:

resource_type (str): The type of resource that conflicts (e.g., “file_path”, “project_endpoint”) identifier (Any): The identifier for this resource existing_instance (Any): The existing instance that conflicts with this one

Returns:

A clear error message explaining the conflict and how to resolve it.

_register_resources()

Register this instance’s resources and check for conflicts.

Raises:

ResourceConflictError: If any resource conflicts with an existing instance.

_cleanup_dead_references(registry: dict[str, weakref.ref])

Remove dead weakref entries from the registry.

Args:

registry (dict[str, weakref.ref]): The registry to clean up.

classmethod get_active_resource_count() → int

Get the number of active resources registered for this class.

Returns:

int: Number of active resource registrations.