Merge pull request #226 from lsst-sqre/tickets/DM-38339

DM-38339: Add support for running a business once

src/mobu/dependencies/context.py

@@ -13,7 +13,7 @@
 from safir.dependencies.gafaelfawr import auth_logger_dependency
 from structlog.stdlib import BoundLogger
 
-from ..factory import ProcessContext
+from ..factory import Factory, ProcessContext
 from ..services.manager import FlockManager
 
 __all__ = [
@@ -36,6 +36,9 @@ class RequestContext:
     manager: FlockManager
     """Global singleton flock manager."""
 
+    factory: Factory
+    """Component factory."""
+
 
 class ContextDependency:
     """Provide a per-request context as a FastAPI dependency.
@@ -61,6 +64,7 @@ async def __call__(
             request=request,
             logger=logger,
             manager=self._process_context.manager,
+            factory=Factory(self._process_context, logger),
         )
 
     @property

src/mobu/factory.py

@@ -7,10 +7,12 @@
 import structlog
 from aiohttp import ClientSession
 from safir.slack.webhook import SlackWebhookClient
-from structlog import BoundLogger
+from structlog.stdlib import BoundLogger
 
 from .config import config
+from .models.solitary import SolitaryConfig
 from .services.manager import FlockManager
+from .services.solitary import Solitary
 
 __all__ = ["Factory", "ProcessContext"]
 
@@ -72,3 +74,18 @@ def create_slack_webhook_client(self) -> SlackWebhookClient | None:
         if not config.alert_hook or config.alert_hook == "None":
             return None
         return SlackWebhookClient(config.alert_hook, "Mobu", self._logger)
+
+    def create_solitary(self, solitary_config: SolitaryConfig) -> Solitary:
+        """Create a runner for a solitary monkey.
+
+        Parameters
+        ----------
+        solitary_config
+            Configuration for the solitary monkey.
+
+        Returns
+        -------
+        Solitary
+            Newly-created solitary manager.
+        """
+        return Solitary(solitary_config, self._context.session, self._logger)

src/mobu/handlers/external.py

@@ -14,6 +14,7 @@
 from ..models.flock import FlockConfig, FlockData, FlockSummary
 from ..models.index import Index
 from ..models.monkey import MonkeyData
+from ..models.solitary import SolitaryConfig, SolitaryResult
 
 external_router = APIRouter()
 """FastAPI router for all external handlers."""
@@ -180,6 +181,26 @@ async def get_flock_summary(
     return context.manager.get_flock(flock).summary()
 
 
+@external_router.post(
+    "/run",
+    response_class=FormattedJSONResponse,
+    response_model=SolitaryResult,
+    response_model_exclude_none=True,
+    response_model_exclude_unset=True,
+    summary="Run monkey business once",
+)
+async def put_run(
+    solitary_config: SolitaryConfig,
+    context: RequestContext = Depends(context_dependency),
+) -> SolitaryResult:
+    context.logger.info(
+        "Running solitary monkey",
+        config=solitary_config.dict(exclude_unset=True),
+    )
+    solitary = context.factory.create_solitary(solitary_config)
+    return await solitary.run()
+
+
 @external_router.get(
     "/summary",
     response_class=FormattedJSONResponse,

src/mobu/models/business/base.py

@@ -37,8 +37,7 @@ class BusinessConfig(BaseModel):
 
     Each type of business must override this class, redefining ``type`` with a
     different literal and ``options`` with a different type and default
-    factory. This base class doubles as the configuration for the
-    `~mobu.services.business.base.Business` base business class.
+    factory.
     """
 
     type: str = Field(..., title="Type of business to run")

src/mobu/models/solitary.py

@@ -0,0 +1,47 @@
+"""Models for running a single instance of a business by itself."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from .business.empty import EmptyLoopConfig
+from .business.jupyterpythonloop import JupyterPythonLoopConfig
+from .business.notebookrunner import NotebookRunnerConfig
+from .business.tapqueryrunner import TAPQueryRunnerConfig
+from .user import User
+
+
+class SolitaryConfig(BaseModel):
+    """Configuration for a solitary monkey.
+
+    This is similar to `~mobu.models.flock.FlockConfig`, but less complex
+    since it can only wrap a single monkey business.
+    """
+
+    user: User = Field(..., title="User to run as")
+
+    scopes: list[str] = Field(
+        ...,
+        title="Token scopes",
+        description="Must include all scopes required to run the business",
+        example=["exec:notebook", "read:tap"],
+    )
+
+    business: (
+        TAPQueryRunnerConfig
+        | NotebookRunnerConfig
+        | JupyterPythonLoopConfig
+        | EmptyLoopConfig
+    ) = Field(..., title="Business to run")
+
+
+class SolitaryResult(BaseModel):
+    """Results from executing a solitary monkey."""
+
+    success: bool = Field(..., title="Whether the business succeeded")
+
+    error: Optional[str] = Field(None, title="Error if the business failed")
+
+    log: str = Field(..., title="Log of the business execution")

src/mobu/services/business/base.py

@@ -151,6 +151,20 @@ async def run(self) -> None:
             if self.stopping:
                 self.control.task_done()
 
+    async def run_once(self) -> None:
+        """The core business logic, run only once.
+
+        Calls `startup`, `execute`, `shutdown`, and `close`.
+        """
+        self.logger.info("Starting up...")
+        try:
+            await self.startup()
+            await self.execute()
+            self.logger.info("Shutting down...")
+            await self.shutdown()
+        finally:
+            await self.close()
+
     async def idle(self) -> None:
         """The idle pause at the end of each loop."""
         self.logger.info("Idling...")

src/mobu/services/monkey.py

@@ -123,6 +123,26 @@ def logfile(self) -> str:
         self._logfile.flush()
         return self._logfile.name
 
+    async def run_once(self) -> str | None:
+        """Run the monkey business once.
+
+        Returns
+        -------
+        str or None
+            Error message on failure, or `None` if the business succeeded.
+        """
+        self._state = MonkeyState.RUNNING
+        error = None
+        try:
+            await self.business.run_once()
+            self._state = MonkeyState.FINISHED
+        except Exception as e:
+            msg = "Exception thrown while doing monkey business"
+            self._logger.exception(msg)
+            error = str(e)
+            self._state = MonkeyState.ERROR
+        return error
+
     async def start(self, scheduler: Scheduler) -> None:
         self._job = await scheduler.spawn(self._runner())
 

src/mobu/services/solitary.py

@@ -0,0 +1,63 @@
+"""Manager for a solitary monkey."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from aiohttp import ClientSession
+from structlog.stdlib import BoundLogger
+
+from ..models.solitary import SolitaryConfig, SolitaryResult
+from ..models.user import AuthenticatedUser
+from .monkey import Monkey
+
+__all__ = ["Solitary"]
+
+
+class Solitary:
+    """Runs a single monkey to completion and reports its results.
+
+    Parameters
+    ----------
+    solitary_config
+        Configuration for the monkey.
+    session
+        HTTP client session.
+    logger
+        Global logger.
+    """
+
+    def __init__(
+        self,
+        solitary_config: SolitaryConfig,
+        session: ClientSession,
+        logger: BoundLogger,
+    ) -> None:
+        self._config = solitary_config
+        self._session = session
+        self._logger = logger
+
+    async def run(self) -> SolitaryResult:
+        """Run the monkey and return its results.
+
+        Returns
+        -------
+        SolitaryResult
+            Result of monkey run.
+        """
+        user = await AuthenticatedUser.create(
+            self._config.user, self._config.scopes, self._session
+        )
+        monkey = Monkey(
+            name=f"solitary-{user.username}",
+            business_config=self._config.business,
+            user=user,
+            session=self._session,
+            logger=self._logger,
+        )
+        error = await monkey.run_once()
+        return SolitaryResult(
+            success=error is None,
+            error=error,
+            log=Path(monkey.logfile()).read_text(),
+        )

tests/handlers/solitary_test.py

@@ -0,0 +1,63 @@
+"""Tests for running a solitary monkey."""
+
+from __future__ import annotations
+
+from unittest.mock import ANY
+
+import pytest
+from aioresponses import aioresponses
+from httpx import AsyncClient
+from safir.testing.slack import MockSlackWebhook
+
+from ..support.gafaelfawr import mock_gafaelfawr
+
+
+@pytest.mark.asyncio
+async def test_run(
+    client: AsyncClient, mock_aioresponses: aioresponses
+) -> None:
+    mock_gafaelfawr(mock_aioresponses)
+
+    r = await client.post(
+        "/mobu/run",
+        json={
+            "user": {"username": "solitary"},
+            "scopes": ["exec:notebook"],
+            "business": {"type": "EmptyLoop"},
+        },
+    )
+    assert r.status_code == 200
+    result = r.json()
+    assert result == {"success": True, "log": ANY}
+    assert "Starting up..." in result["log"]
+    assert "Shutting down..." in result["log"]
+
+
+@pytest.mark.asyncio
+async def test_error(
+    client: AsyncClient,
+    slack: MockSlackWebhook,
+    mock_aioresponses: aioresponses,
+) -> None:
+    mock_gafaelfawr(mock_aioresponses)
+
+    r = await client.post(
+        "/mobu/run",
+        json={
+            "user": {"username": "solitary"},
+            "scopes": ["exec:notebook"],
+            "business": {
+                "type": "JupyterPythonLoop",
+                "options": {
+                    "code": 'raise Exception("some error")',
+                    "spawn_settle_time": 0,
+                },
+            },
+        },
+    )
+    assert r.status_code == 200
+    result = r.json()
+    assert result == {"success": False, "error": ANY, "log": ANY}
+    assert "solitary: running code 'raise Exception" in result["error"]
+    assert "Exception: some error\n" in result["error"]
+    assert "Exception: some error" in result["log"]