orchestrata_ultimate.core.base.service

Reference › orchestrata_ultimate › core › base › service

The base class for services, implementing the IComponent lifecycle from core.protocols. BaseService stores an optional configuration dictionary, creates a logger named after the concrete subclass, and holds a threading.Lock for free-threaded safety. Its lifecycle methods are defaults to override: initialize stores the configuration, start and stop do nothing, get_health returns HEALTHY, and get_metadata builds ComponentMetadata from the class name.

At a glance

Class diagram

Dependency diagram

API

Classes

BaseService

Bases: IComponent · Line 16

Base class for all services.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/base/service.py

"""BaseService — service base with config, logging, and IComponent lifecycle."""

import logging
import threading
import uuid
from typing import Any

from orchestrata_ultimate.core.protocols import (
    ComponentMetadata,
    ComponentType,
    HealthStatus,
    IComponent,
)


class BaseService(IComponent):
    """Base class for all services.

    Provides config injection, auto-named logger, threading.Lock,
    and concrete no-op IComponent lifecycle defaults.

    Attributes:
        _config: Service configuration dict (or None).
        _logger: Logger named by concrete subclass.
        _lock: Threading lock for 3.14t safety.
    """

    def __init__(
        self,
        config: dict[str, Any] | None = None,
    ) -> None:
        """Initialize service with optional config.

        Args:
            config: Configuration dict (optional).
        """
        self._config = config or {}
        self._logger = logging.getLogger(self.__class__.__qualname__)
        self._lock = threading.Lock()

    async def initialize(
        self,
        config: dict[str, Any],
    ) -> None:
        """IComponent: store config. Override for custom init.

        Args:
            config: Configuration dictionary.
        """
        self._config = config

    async def start(self) -> None:
        """IComponent: no-op. Override for startup logic."""

    async def stop(self) -> None:
        """IComponent: no-op. Override for shutdown logic."""

    async def get_health(self) -> HealthStatus:
        """IComponent: returns HEALTHY. Override for real checks.

        Returns:
            HealthStatus.HEALTHY by default.
        """
        return HealthStatus.HEALTHY

    def get_metadata(self) -> ComponentMetadata:
        """IComponent: auto-generated metadata from class name.

        Returns:
            ComponentMetadata with auto-populated fields.
        """
        return ComponentMetadata(
            id=str(uuid.uuid4()),
            name=self.__class__.__qualname__,
            version="0.1.0",
            component_type=ComponentType.SERVICE,
        )