Skip to main content

IEngine

What it is​

  • Defines the core engine interface/container for:
    • A set of shared services (IEngine.Services)
    • A registry of loaded modules (modules)
  • Provides a small wiring mechanism to let services reference each other when they implement a known protocol (ServicesAware).

Public API​

ServicesAware (Protocol)​

  • set_services(self, services: IEngine.Services) -> None
    • Implement on a service to receive the engine’s Services container.

IEngine​

  • services: IEngine.Services (property)
    • Returns the engine service container.
  • modules: Dict[str, BaseModule] (property)
    • Returns the module registry keyed by module name.

IEngine.Services​

Container for optional engine services, with guarded accessors.

Constructor

  • Services(object_storage=None, triple_store=None, vector_store=None, secret=None, bus=None, kv=None)

Service accessors (raise AssertionError if missing)

  • kv -> KeyValueService
  • object_storage -> ObjectStorageService
  • triple_store -> TripleStoreService
  • vector_store -> VectorStoreService
  • secret -> Secret
  • bus -> BusService

Other methods/properties

  • triple_store_available() -> bool
    • Returns True if a triple store service was provided.
  • all -> List[Union[...]]
    • Returns a list of all service instances (including None entries) in a fixed order.
  • wire_services() -> None
    • For each non-None service in all, if it implements ServicesAware, calls service.set_services(self).

Configuration/Dependencies​

  • Depends on service types from naas_abi_core.services.*:
    • ObjectStorageService, TripleStoreService, VectorStoreService, Secret, BusService, KeyValueService
  • BaseModule is only imported for typing (TYPE_CHECKING) from naas_abi_core.module.Module.

Usage​

from naas_abi_core.engine.IEngine import IEngine, ServicesAware

class MyService(ServicesAware):
    def set_services(self, services: IEngine.Services) -> None:
        # Can now access other services via `services`
        if services.triple_store_available():
            _ = services.triple_store

# Provide service instances as available in your environment
services = IEngine.Services(
    object_storage=None,
    triple_store=None,
    vector_store=None,
    secret=None,
    bus=None,
    kv=None,
)

# Wire cross-service references (no-op unless services implement ServicesAware)
services.wire_services()

Caveats​

  • Accessing a missing service via its property (e.g., services.kv) raises AssertionError with a clear message.
  • wire_services() only calls set_services for services that are runtime-checkable instances of ServicesAware.