plugin_author_guide.md

OpenFetch - Plugin Author Guide

OpenFetch is designed to be highly extensible. You can easily add support for your own datasets by writing custom connectors (plugins).

This guide details how to implement a connector and register it in the registry.

---

Connector Interface

Every connector must subclass BaseConnector located in openfetch/core/connector.py.

Here is the template for a custom connector:

from typing import Any, Dict, List
from openfetch.core.connector import BaseConnector
from openfetch.core.models import FetchResult
from openfetch.core.fetcher import download_file_async

class MyDatasetConnector(BaseConnector):
    @property
    def name(self) -> str:
        """Returns the user-friendly name of the custom dataset."""
        return "My Dataset"

    @property
    def description(self) -> str:
        """Returns a brief description of what this dataset provides."""
        return "A description of what my dataset provides."

    @property
    def category(self) -> str:
        """Categorizes the connector under one of the default groups (e.g., General, Weather / Models)."""
        return "General"  # Options: 'Weather / Models', 'Radar / Satellite', 'Observations', 'General'

    @property
    def aliases(self) -> List[str]:
        """Provides CLI aliases so users can invoke the connector using shorter keywords."""
        return ["mydataset", "mydata"]

    @property
    def quick_actions(self) -> Dict[str, Dict[str, Any]]:
        """Defines default preset queries for quick button selection in the UI."""
        return {
            "default_action": {
                "label": "Get Default Action",
                "params": {"variable": "temp", "limit": 10}
            }
        }

    @property
    def parameters_schema(self) -> Dict[str, Dict[str, Any]]:
        """Defines the metadata, types, and descriptions of the parameters for CLI validation and UI form generation."""
        return {
            "region": {
                "type": "str",
                "default": "native",
                "description": "Geographic model domain or plot region.",
                "option_role": "region",
                "options": [
                    {"value": "native", "label": "Native model domain"},
                    {"value": "conus", "label": "CONUS"}
                ]
            },
            "variable": {
                "type": "str",
                "default": "temp",
                "description": "Select grid variable",
                "option_role": "products"
            },
            "limit": {
                "type": "int",
                "default": 10,
                "description": "Output row limit"
            }
        }

    async def detect_latest(self) -> Dict[str, Any]:
        """Performs date/time math or quick network calls to check and return the latest available runs."""
        return {"variable": "temp", "limit": 10}

    def build_request(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """Converts user parameters into remote endpoint HTTP request configurations (URL, parameters, etc.)."""
        return {
            "url": "https://api.mydata.org/v1/query",
            "params": {
                "var": params.get("variable", "temp"),
                "limit": int(params.get("limit", 10))
            }
        }

    def preview(self, params: Dict[str, Any]) -> str:
        """Generates a text overview shown in the right-hand panel of the TUI before fetching."""
        req = self.build_request(params)
        return f"Fetch MyDataset.\nTarget: {req['url']}?{req['params']}"

    def cache_key(self, params: Dict[str, Any]) -> str:
        """Generates a unique, stable cache key based on query parameters."""
        var = params.get("variable", "temp")
        limit = params.get("limit", 10)
        return f"mydata_{var}_l{limit}"

    async def fetch(self, params: Dict[str, Any], cache_store: Any) -> FetchResult:
        """Main execution function triggered on fetch. Handles cache hit checks and schedules async download."""
        ckey = self.cache_key(params)
        cached_entry = cache_store.get(ckey)

        cli_command = f"openfetch fetch mydata {params.get('variable')}"

        if cached_entry:
            file_path = cached_entry["file_path"]
            return FetchResult(
                connector=self.name,
                success=True,
                file_path=file_path,
                size_bytes=cached_entry["size_bytes"],
                message="Retrieved from cache.",
                command_copied=cli_command,
                data_preview=self.format_result(file_path)
            )

        # Execute remote download
        req_info = self.build_request(params)
        dest_file = cache_store.cache_dir / f"{ckey}.json"

        try:
            bytes_written = await download_file_async(
                url=req_info["url"],
                dest_path=str(dest_file),
                params=req_info["params"]
            )

            # Cache locally (TTL: 1 hour)
            cache_store.set(
                connector=self.name,
                query_key=ckey,
                file_path=str(dest_file),
                size_bytes=bytes_written,
                params=params,
                ttl_seconds=3600
            )

            return FetchResult(
                connector=self.name,
                success=True,
                file_path=str(dest_file),
                size_bytes=bytes_written,
                message="Successfully downloaded dataset.",
                command_copied=cli_command,
                data_preview=self.format_result(str(dest_file))
            )
        except Exception as e:
            return FetchResult(
                connector=self.name,
                success=False,
                message=f"Fetch failed: {str(e)}",
                command_copied=cli_command
            )

    def format_result(self, file_path: str) -> str:
        """Reads file from disk and parses it into a friendly console text block representation."""
        import os
        if not os.path.exists(file_path):
            return "File missing."
        return f"Successfully saved to: {file_path}"

---

Registering Your Connector

Once you have written your connector subclass:

1. Save it under openfetch/connectors/my_dataset/connector.py.
2. Open openfetch/core/registry.py.
3. Import your connector at the top of the initialize_registry function:

   from openfetch.connectors.my_dataset.connector import MyDatasetConnector

4. Register the instance in initialize_registry:

   registry.register(MyDatasetConnector())

5. Test it! It will automatically appear in both the openfetch interactive TUI sidebar and the openfetch fetch CLI completions!