Create a Runtime Plugin

plugins/warehouse/
├── plugin.yaml
├── plugin.py
├── pyproject.toml       optional SDK dependencies
├── lib/                 private plugin helpers
└── extensions/          declared Harnest lifecycle contributions

import os

from harnest.plugins import Plugin, PluginContext
from warehouse_sdk import WarehouseClient


class WarehouseContext(PluginContext):
    """Expose bounded warehouse operations during one invocation."""

    __slots__ = ("_owner",)

    def __init__(self, plugin_name: str, owner: "Warehouse") -> None:
        """Bind the revocable context to its application-owned plugin."""

        super().__init__(plugin_name)
        self._owner = owner

    async def query(self, statement: str) -> list[dict]:
        """Run one query while the Harnest invocation remains active."""

        self._require_active()
        return await self._owner._query(statement)


class Warehouse(Plugin[WarehouseContext]):
    """Own one warehouse SDK client for the application lifetime."""

    def __init__(self) -> None:
        """Defer all external connection work until runtime startup."""

        self._client = None

    async def start(self, _context) -> None:
        """Connect after dependency-ordered application startup begins."""

        self._client = await WarehouseClient.connect(
            endpoint=os.environ["WAREHOUSE_ENDPOINT"],
        )

    async def stop(self) -> None:
        """Release the application-owned SDK client."""

        # Shutdown must also be safe after a partial startup failure.
        if self._client is not None:
            await self._client.close()
            self._client = None

    def create_context(self, base: PluginContext) -> WarehouseContext:
        """Create one typed, revocable view for the active invocation."""

        return WarehouseContext(base.plugin_name, self)

    async def query(self, statement: str) -> list[dict]:
        """Route public calls through the current invocation context."""

        return await self.context.query(statement)

    async def _query(self, statement: str) -> list[dict]:
        """Cross the private SDK boundary after startup validation."""

        # Calls fail closed if they escape the managed application lifetime.
        if self._client is None:
            raise RuntimeError("Warehouse plugin is not started")
        return await self._client.query(statement)


plugin = Warehouse()

requires:
  plugins: [credentials]