ad-sdl/madsci/madsci-managers

from madsci.common.types.manager_types import ManagerSettings
from madsci.common.types.base_types import prefixed_alias_generator, prefixed_model_validator
class MyManagerSettings(ManagerSettings):
    model_config = SettingsConfigDict(
        env_prefix="MYMANAGER_",
        alias_generator=prefixed_alias_generator("mymanager"),
        populate_by_name=True,
    )
    _accept_prefixed_keys = prefixed_model_validator("mymanager")
    server_url: AnyUrl = Field(default="http://localhost:8007/")
    document_db_url: AnyUrl = Field(default="mongodb://localhost:27017/")
    database_name: str = Field(default="madsci_mymanager")

from madsci.common.manager_base import AbstractManagerBase
class MyManager(AbstractManagerBase[MyManagerSettings]):
    SETTINGS_CLASS = MyManagerSettings
    def initialize(self, **kwargs):
        """Called during __init__ after base setup. Set up DB connections, etc."""
        self._document_handler = kwargs.get("document_handler") or PyDocumentStorageHandler(
            url=str(self.settings.document_db_url),
            database_name=self.settings.database_name,
        )
    def get_health(self) -> MyManagerHealth:
        """Override to add DB connectivity checks."""
        try:
            connected = self._document_handler.ping()
            return MyManagerHealth(healthy=connected, db_connected=connected)
        except Exception as e:
            return MyManagerHealth(healthy=False, description=str(e))
    @get("/items")
    def list_items(self) -> list[Item]:
        """Custom endpoint using classy_fastapi @get/@post decorators."""
        ...
    @post("/items")
    def create_item(self, item: Item) -> Item:
        ...

from madsci.common.utils import create_http_session
from pydantic import AnyUrl
class MyManagerClient:
    def __init__(self, server_url: AnyUrl, config=None, event_client=None):
        self.server_url = str(server_url)
        self._session = None  # Lazy HTTP session
    def list_items(self, timeout=None) -> list[Item]:
        response = self._get("/items", timeout=timeout)
        return [Item(**item) for item in response.json()]
    def create_item(self, item: Item, timeout=None) -> Item:
        response = self._post("/items", json=item.model_dump(), timeout=timeout)
        return Item(**response.json())
    def close(self):
        """Release HTTP session resources."""
        if self._session:
            self._session.close()

# settings.yaml (shared across managers)
event_server_url: http://localhost:8001/
event_database_name: madsci_events
resource_postgres_db_url: postgresql://localhost/resources
workcell_cache_url: redis://localhost:6379/

# In code, use unprefixed names:
self.settings.server_url       # "http://localhost:8001/"
self.settings.database_name    # "madsci_events"

# Type-based:
password: SecretStr
# Metadata-based:
api_key: str = Field(json_schema_extra={"secret": True})
# Safe export (redacts secrets):
settings.model_dump_safe()
# Settings endpoint uses:
get_settings_export(include_secrets=False)  # Default

# Logging (always works, even without Event Manager)
self.logger.info("Operation started", event_type=EventType.WORKFLOW_START, workflow_id=wf_id)
self.logger.warning("Threshold exceeded", value=42.5)
self.logger.error("Connection failed", error=str(e))
# HTTP client (requires Event Manager)
events = self.event_client.get_events(number=10)
event = self.event_client.get_event(event_id)

# Good: kwargs are queryable
self.logger.info("Step completed", step_index=3, duration_ms=150)
# Bad: f-string data is not queryable
self.logger.info(f"Step 3 completed in 150ms")

def get_health(self) -> MyManagerHealth:
    health = MyManagerHealth(healthy=True)
    try:
        health.db_connected = self._document_handler.ping()
        health.total_items = self._get_count()
    except Exception as e:
        health.healthy = False
        health.description = f"Database error: {e}"
    return health

from madsci.common.db_handlers import (
    InMemoryDocumentStorageHandler,
    InMemoryCacheHandler,
    SQLiteHandler,
)
def test_event_manager():
    manager = EventManager(
        settings=EventManagerSettings(enable_registry_resolution=False),
        document_handler=InMemoryDocumentStorageHandler(),
    )
    app = manager.create_server()
    client = TestClient(app)
    # Test endpoints
    response = client.get("/health")
    assert response.status_code == 200
    assert response.json()["healthy"] is True
def test_workcell_manager():
    manager = WorkcellManager(
        settings=WorkcellManagerSettings(enable_registry_resolution=False),
        document_handler=InMemoryDocumentStorageHandler(),
        cache_handler=InMemoryCacheHandler(),
    )
    ...
def test_resource_manager():
    manager = ResourceManager(
        settings=ResourceManagerSettings(enable_registry_resolution=False),
        postgres_handler=SQLiteHandler(),
    )
    ...

EVENT_OTEL_ENABLED=true
EVENT_OTEL_SERVICE_NAME="madsci.event"
EVENT_OTEL_EXPORTER="otlp"           # or "console"
EVENT_OTEL_ENDPOINT="http://localhost:4317"
EVENT_OTEL_PROTOCOL="grpc"           # or "http"

with self.span("process_data", attributes={"data.size": 100}) as span:
    result = process(data)
    span.set_attribute("result.count", len(result))