Create a Harnest Extension

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

import os

from harnest.extensions import Extension, ExtensionContext
from warehouse_sdk import WarehouseClient


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

    __slots__ = ("_owner",)

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

        super().__init__(extension_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(Extension[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: ExtensionContext) -> WarehouseContext:
        """Create one typed, revocable view for the active invocation."""

        return WarehouseContext(base.extension_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 extension is not started")
        return await self._client.query(statement)


extension = Warehouse()

requires:
  extensions: [credentials]