Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,10 @@ python = "^3.12"
# NOTE: Tags are acceptable for development; consider commit SHAs for production deployments.
#
# omnibase-core v0.6.2: Latest release with updated models and enums.
# Note (OMN-1257): In omnibase_core v0.6.2+, ModelONEXContainer.service_registry
# returns None when enable_service_registry=False was passed to constructor or
# ServiceRegistry module is unavailable. container_wiring.py now raises
# ServiceRegistryUnavailableError with actionable hints instead of cryptic AttributeError.
omnibase-core = "^0.6.2"
omnibase-spi = "^0.4.1"

Expand Down
6 changes: 3 additions & 3 deletions scripts/dlq_replay.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@
from pydantic import BaseModel, Field, ValidationInfo, field_validator

from omnibase_infra.dlq import (
DLQTrackingService,
DLQReplayTracker,
EnumReplayStatus,
ModelDlqReplayRecord,
ModelDlqTrackingConfig,
Expand Down Expand Up @@ -1023,7 +1023,7 @@ def __init__(self, config: ModelReplayConfig) -> None:
self.consumer = DLQConsumer(config)
self.producer = DLQProducer(config)
self.results: list[ModelReplayResult] = []
self._tracking_service: DLQTrackingService | None = None
self._tracking_service: DLQReplayTracker | None = None

@property
def is_tracking_enabled(self) -> bool:
Expand All @@ -1050,7 +1050,7 @@ async def start(self) -> None:
dsn = self.config.build_tracking_dsn()
if dsn:
tracking_config = ModelDlqTrackingConfig(dsn=dsn)
self._tracking_service = DLQTrackingService(tracking_config)
self._tracking_service = DLQReplayTracker(tracking_config)
try:
await self._tracking_service.initialize()
logger.info("DLQ tracking service initialized")
Expand Down
9 changes: 2 additions & 7 deletions src/omnibase_infra/dlq/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
Tracker:
- DLQReplayTracker: PostgreSQL tracker for replay history
- ServiceDlqTracking: ONEX naming convention alias
- DLQTrackingService: Backwards compatibility alias

Example - Recording Replay Attempts:
>>> from omnibase_infra.dlq import (
Expand Down Expand Up @@ -82,19 +81,15 @@
ServiceDlqTracking,
)

# Backwards compatibility alias
DLQTrackingService = DLQReplayTracker

__all__: list[str] = [
# Constants
"PATTERN_TABLE_NAME", # Regex pattern string for table name validation
"REGEX_TABLE_NAME", # Pre-compiled regex for runtime validation
# Tracker (multiple names for flexibility)
# Tracker
"DLQReplayTracker", # Primary class name
"DLQTrackingService", # Backwards compatibility alias
"ServiceDlqTracking", # ONEX naming convention (service_<name>.py → Service<Name>)
# Models
"EnumReplayStatus",
"ModelDlqReplayRecord",
"ModelDlqTrackingConfig",
"ServiceDlqTracking", # ONEX naming convention (service_<name>.py → Service<Name>)
]
2 changes: 2 additions & 0 deletions src/omnibase_infra/errors/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@
ContainerValidationError,
ContainerWiringError,
ServiceRegistrationError,
ServiceRegistryUnavailableError,
ServiceResolutionError,
)
from omnibase_infra.errors.error_infra import (
Expand Down Expand Up @@ -128,6 +129,7 @@
"RuntimeHostError",
"SecretResolutionError",
"ServiceRegistrationError",
"ServiceRegistryUnavailableError",
"ServiceResolutionError",
"UnknownHandlerTypeError",
]
60 changes: 60 additions & 0 deletions src/omnibase_infra/errors/error_container_wiring.py
Original file line number Diff line number Diff line change
Expand Up @@ -176,9 +176,69 @@ def __init__(
)


class ServiceRegistryUnavailableError(ContainerValidationError):
"""Raised when container.service_registry is None.

This error indicates that the ModelONEXContainer was initialized
without a service registry, either because:
- enable_service_registry=False was passed
- The ServiceRegistry module is not installed/available
- Container initialization failed silently

This is a distinct error from generic ContainerValidationError to provide
clearer diagnostics and actionable error messages for service_registry issues.

Example:
>>> raise ServiceRegistryUnavailableError(
... "Container service_registry is None",
... operation="register PolicyRegistry",
... hint="Check that enable_service_registry=True",
... )
"""

def __init__(
self,
message: str,
*,
operation: str | None = None,
hint: str | None = None,
context: ModelInfraErrorContext | None = None,
**extra_context: object,
) -> None:
"""Initialize ServiceRegistryUnavailableError.

Args:
message: Primary error message.
operation: The operation that was attempted (e.g., 'register_instance').
hint: Actionable hint for fixing the issue.
context: Bundled infrastructure context.
**extra_context: Additional context information.
"""
# Build full message with operation and hint
full_message = message
if operation:
full_message = f"{message} (operation: {operation})"
extra_context["operation"] = operation

default_hint = (
"Ensure ModelONEXContainer is created with enable_service_registry=True "
"and that the ServiceRegistry module is installed."
)
actual_hint = hint or default_hint
full_message = f"{full_message}\nHint: {actual_hint}"
extra_context["hint"] = actual_hint

super().__init__(
message=full_message,
context=context,
**extra_context,
)


__all__ = [
"ContainerValidationError",
"ContainerWiringError",
"ServiceRegistrationError",
"ServiceRegistryUnavailableError",
"ServiceResolutionError",
]
103 changes: 25 additions & 78 deletions src/omnibase_infra/event_bus/kafka_event_bus.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,13 +133,18 @@
Usage:
```python
from omnibase_infra.event_bus.kafka_event_bus import KafkaEventBus
from omnibase_infra.event_bus.models.config import ModelKafkaEventBusConfig

# Option 1: Use defaults with environment variable overrides
bus = KafkaEventBus.default()
await bus.start()

# Option 2: Explicit configuration
bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev")
# Option 2: Explicit configuration via config model
config = ModelKafkaEventBusConfig(
bootstrap_servers="kafka:9092",
environment="dev",
)
bus = KafkaEventBus(config=config)
await bus.start()

# Subscribe to a topic
Expand Down Expand Up @@ -243,7 +248,11 @@ class KafkaEventBus(MixinAsyncCircuitBreaker):

Example:
```python
bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev")
config = ModelKafkaEventBusConfig(
bootstrap_servers="kafka:9092",
environment="dev",
)
bus = KafkaEventBus(config=config)
await bus.start()

# Subscribe
Expand All @@ -263,45 +272,12 @@ async def handler(msg):
def __init__(
self,
config: ModelKafkaEventBusConfig | None = None,
# Backwards compatibility parameters (override config if provided)
bootstrap_servers: str | None = None,
environment: str | None = None,
group: str | None = None,
timeout_seconds: int | None = None,
max_retry_attempts: int | None = None,
retry_backoff_base: float | None = None,
circuit_breaker_threshold: int | None = None,
circuit_breaker_reset_timeout: float | None = None,
) -> None:
"""Initialize the Kafka event bus.

Design Note:
This __init__ method intentionally accepts 10 parameters (1 config + 8 overrides + self)
to maintain backwards compatibility while transitioning to config-driven initialization.
The parameters follow the ONEX config pattern:
- Primary: `config` parameter (ModelKafkaEventBusConfig) for modern usage
- Overrides: 8 optional parameters for backwards compatibility

This approach allows gradual migration from direct parameters to config objects
without breaking existing code. Recommended usage is factory methods:
- KafkaEventBus.default() for defaults with env var overrides
- KafkaEventBus.from_config(config) for config-driven initialization
- KafkaEventBus.from_yaml(path) for YAML-based configuration

The parameter count is acceptable for infrastructure components supporting
multiple initialization patterns during deprecation periods.

Args:
config: Configuration model containing all settings. If not provided,
defaults are used with environment variable overrides.
bootstrap_servers: Override bootstrap servers from config
environment: Override environment identifier from config
group: Override consumer group identifier from config
timeout_seconds: Override timeout from config
max_retry_attempts: Override max retry attempts from config
retry_backoff_base: Override retry backoff base from config
circuit_breaker_threshold: Override circuit breaker threshold from config
circuit_breaker_reset_timeout: Override circuit breaker reset timeout from config

Raises:
ProtocolConfigurationError: If circuit_breaker_threshold is not a positive integer
Expand All @@ -318,9 +294,6 @@ def __init__(
# Using factory methods
bus = KafkaEventBus.default()
bus = KafkaEventBus.from_yaml(Path("kafka.yaml"))

# Backwards compatible direct parameters
bus = KafkaEventBus(bootstrap_servers="kafka:9092", environment="dev")
```
"""
# Use provided config or create default with environment overrides
Expand All @@ -330,59 +303,33 @@ def __init__(
# Store config reference
self._config = config

# Apply parameter overrides for backwards compatibility
self._bootstrap_servers = (
bootstrap_servers
if bootstrap_servers is not None
else config.bootstrap_servers
)
self._environment = (
environment if environment is not None else config.environment
)
self._group = group if group is not None else config.group
self._timeout_seconds = (
timeout_seconds if timeout_seconds is not None else config.timeout_seconds
)
self._max_retry_attempts = (
max_retry_attempts
if max_retry_attempts is not None
else config.max_retry_attempts
)
self._retry_backoff_base = (
retry_backoff_base
if retry_backoff_base is not None
else config.retry_backoff_base
)
# Apply config values
self._bootstrap_servers = config.bootstrap_servers
self._environment = config.environment
self._group = config.group
self._timeout_seconds = config.timeout_seconds
self._max_retry_attempts = config.max_retry_attempts
self._retry_backoff_base = config.retry_backoff_base

# Circuit breaker configuration with override support
effective_threshold = (
circuit_breaker_threshold
if circuit_breaker_threshold is not None
else config.circuit_breaker_threshold
)
if effective_threshold < 1:
# Circuit breaker configuration
if config.circuit_breaker_threshold < 1:
context = ModelInfraErrorContext(
transport_type=EnumInfraTransportType.KAFKA,
operation="init",
target_name="kafka_event_bus",
correlation_id=uuid4(),
)
raise ProtocolConfigurationError(
f"circuit_breaker_threshold must be a positive integer, got {effective_threshold}",
f"circuit_breaker_threshold must be a positive integer, got {config.circuit_breaker_threshold}",
context=context,
parameter="circuit_breaker_threshold",
value=effective_threshold,
value=config.circuit_breaker_threshold,
)
effective_reset_timeout = (
circuit_breaker_reset_timeout
if circuit_breaker_reset_timeout is not None
else config.circuit_breaker_reset_timeout
)

# Initialize circuit breaker mixin
self._init_circuit_breaker(
threshold=effective_threshold,
reset_timeout=effective_reset_timeout,
threshold=config.circuit_breaker_threshold,
reset_timeout=config.circuit_breaker_reset_timeout,
service_name=f"kafka.{self._environment}",
transport_type=EnumInfraTransportType.KAFKA,
)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -189,7 +189,7 @@ class ModelKafkaEventBusConfig(BaseModel):
retry_backoff_base: float = Field(
default=1.0,
description="Base delay in seconds for exponential backoff",
ge=0.1,
ge=0.001, # Allow very short backoffs for testing (minimum 1ms)
le=60.0,
)

Expand All @@ -203,7 +203,7 @@ class ModelKafkaEventBusConfig(BaseModel):
circuit_breaker_reset_timeout: float = Field(
default=30.0,
description="Seconds before circuit breaker resets to half-open state",
ge=1.0,
ge=0.01, # Allow short timeouts for testing (minimum 10ms)
le=3600.0,
)

Expand Down
8 changes: 0 additions & 8 deletions src/omnibase_infra/idempotency/models/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,17 +7,11 @@

Exports:
ModelIdempotencyStoreHealthCheckResult: Result of an idempotency store health check
ModelHealthCheckResult: Backward compatibility alias for ModelIdempotencyStoreHealthCheckResult
ModelIdempotencyRecord: Represents a stored idempotency record
ModelIdempotencyCheckResult: Result of an idempotency check operation
ModelIdempotencyGuardConfig: Configuration for idempotency guard middleware
ModelIdempotencyStoreMetrics: Metrics for store observability
ModelPostgresIdempotencyStoreConfig: Configuration for PostgreSQL store

.. versionchanged:: 0.7.0
Renamed ModelHealthCheckResult to ModelIdempotencyStoreHealthCheckResult
to avoid naming conflicts with runtime.models.ModelHealthCheckResult.
The old name is preserved as a backward compatibility alias.
"""

from omnibase_infra.idempotency.models.model_idempotency_check_result import (
Expand All @@ -30,7 +24,6 @@
ModelIdempotencyRecord,
)
from omnibase_infra.idempotency.models.model_idempotency_store_health_check_result import (
ModelHealthCheckResult, # Backward compatibility alias
ModelIdempotencyStoreHealthCheckResult,
)
from omnibase_infra.idempotency.models.model_idempotency_store_metrics import (
Expand All @@ -41,7 +34,6 @@
)

__all__ = [
"ModelHealthCheckResult", # Backward compatibility alias
"ModelIdempotencyCheckResult",
"ModelIdempotencyGuardConfig",
"ModelIdempotencyRecord",
Expand Down
Loading
Loading