Xpublish

https://xpublish.readthedocs.io/en/0.4.1/
Admin
Xpublish is a Python library that provides a quick and extendable way to serve Xarray datasets over...
Tokens:31,771
Snippets:169
Trust Score:8.3
Update:1 year ago
Show doc for...
Tokens:
Raw
### Build Documentation Locally (Shell)

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/contributing

Commands to build the Xpublish documentation locally. Requires a development environment setup and specific dependencies to be installed.

```shell
$ conda activate xpublish-dev
$ cd docs
$ pip install -r requirements.txt
$ make html
```

--------------------------------

### Setup Plugin and Dataset Routers (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Sets up the plugin and dataset routers. This method needs to be executed after both the dataset and plugin setup is complete to ensure all components are correctly configured. Dependencies include `APIRouter` and `Optional`.

```python
def _init_routers(self, dataset_routers: Optional[APIRouter]) -> None:
    """Setup plugin and dataset routers. Needs to run after dataset and plugin setup."""
```

--------------------------------

### Install Xpublish via Pip

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

Installs the Xpublish package using the pip package manager. This is a standard Python package installation method.

```bash
python -m pip install xpublish
```

--------------------------------

### Install and Use Pre-commit Hooks (Shell)

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/contributing

Steps to install the pre-commit framework and its hooks for the Xpublish repository. Pre-commit automates code style and formatting checks before commits.

```shell
$ pip install --user pre-commit
$ pre-commit install
```

--------------------------------

### Install Xpublish via Conda

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

Installs the Xpublish package using the conda package manager from the conda-forge channel. Ensure you have conda installed and configured.

```bash
conda install -c conda-forge xpublish
```

--------------------------------

### Setup and Load Plugins (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Initializes the plugin manager and loads plugins. If no plugins are explicitly provided, it loads default plugins. It then registers any provided plugins and adds their hook specifications. Dependencies include `pluggy`, `Optional`, and `Dict`.

```python
def setup_plugins(
    self,
    plugins: Optional[Dict[str, Plugin]] = None,
) -> None:
    """Initialize and load plugins from entry_points unless explicitly provided.

    Args:
        plugins: A dictionary of initialized plugins. If provided,
            then the automatic loading of plugins is disabled. Providing an
            empty dictionary will also disable automatic loading of plugins.
    """
    if plugins is None:
        plugins = load_default_plugins()

    self.pm = pluggy.PluginManager('xpublish')
    self.pm.add_hookspecs(PluginSpec)

    for name, plugin in plugins.items():
        self.pm.register(plugin, name=name)

    for hookspec in self.pm.hook.register_hookspec():
        self.pm.add_hookspecs(hookspec)
```

--------------------------------

### Implement an App Router Plugin with FastAPI

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This example demonstrates how to implement an `app_router` hook for an Xpublish plugin. It uses FastAPI's `APIRouter` to define a new GET endpoint '/hello' which returns 'world'. The implementation method is marked with the `@hookimpl` decorator.

```python
from xpublish import Plugin, hookimpl
from fastapi import APIRouter

class HelloWorldPlugin(Plugin):
    name: str = "hello_world"

    @hookimpl
    def app_router(self):
        router = APIRouter()

        @router.get("/hello")
        def get_hello():
            return "world"

        return router
```

--------------------------------

### Xpublish Plugin Registration Example

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-router-plugin

This Python code demonstrates how to register a custom Xpublish plugin directly with a `SingleDatasetRest` instance. It first opens an xarray dataset and then initializes the `SingleDatasetRest` before registering the plugin.

```python
import xarray as xr

from xpublish import Plugin, SingleDatasetRest, hookimpl


# Assume MeanPlugin is defined as above

ds = xr.tutorial.open_dataset(
    'air_temperature',
    chunks={'lat': 5, 'lon': 5},
)

rest = SingleDatasetRest(ds)
rest.register_plugin(MeanPlugin())
rest.serve()

```

--------------------------------

### Serve Xarray Dataset with Xpublish

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/introduction

Provides examples of how to serve a published Xarray Dataset using either the `rest.serve()` method or the `ds.rest.serve()` accessor.

```python
rest.serve()

# or

ds.rest.serve()
```

--------------------------------

### Initialize Xpublish Rest with Plugins

Source: https://xpublish.readthedocs.io/en/0.4.1/user-guide/plugins

This code shows how to initialize the `xpublish.Rest` application and provide plugins directly during initialization. The `plugins` argument takes a dictionary where keys are plugin names and values are plugin instances. This method of providing plugins may prevent automatic loading of other installed plugins.

```python
from xpublish import Rest

rest = Rest(datasets, plugins={"hello-world": HelloWorldPlugin()})


```

--------------------------------

### Install Development Dependencies (Shell)

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/contributing

Commands to set up a conda environment and install necessary development dependencies for Xpublish, including the project in editable mode.

```shell
$ conda create -n xpublish-dev
$ conda activate xpublish-dev
$ pip install -r dev-requirements.txt
$ pip install --no-deps -e .
```

--------------------------------

### Load Plugins with Entry Points in TOML and Python

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

This example demonstrates how to configure xpublish plugins for automatic discovery using Python entry points defined in a `pyproject.toml` file. It shows how to specify a plugin name and its corresponding module path. The Python code then shows how to automatically load these plugins or explicitly provide them to the `Rest` instance, including disabling all plugins.

```toml
# pyproject.toml
[project.entry-points."xpublish.plugin"]
mean_plugin = "my_package.plugins:MeanPlugin"
```

```python
import xarray as xr
from xpublish import Rest

# Plugins auto-load from entry points if not explicitly provided
ds = xr.tutorial.open_dataset("air_temperature")
rest = Rest({"air": ds})  # Automatically loads installed plugins
rest.serve()

# To disable auto-loading and use explicit plugins only
from my_package.plugins import MeanPlugin
rest = Rest({"air": ds}, plugins={"mean": MeanPlugin()})
rest.serve()

# To disable all plugins
rest = Rest({"air": ds}, plugins={})
rest.serve()
```

--------------------------------

### Query Dataset Information via REST API

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

Provides examples of using the `requests` library to query dataset information from an Xpublish server. This includes fetching dataset info, listing available datasets, retrieving variable keys, and getting the full dataset as a dictionary.

```python
import requests

# Get dataset info (for single dataset server)
response = requests.get("http://0.0.0.0:9000/info")
info = response.json()

# For multiple dataset server, use dataset_id prefix
response = requests.get("http://0.0.0.0:9000/datasets/air_temperature/info")
info = response.json()

# List all datasets
response = requests.get("http://0.0.0.0:9000/datasets")
dataset_ids = response.json()

# Get variable keys
response = requests.get("http://0.0.0.0:9000/datasets/air_temperature/keys")
variables = response.json()

# Get full dataset as dictionary
response = requests.get("http://0.0.0.0:9000/datasets/air_temperature/dict")
data_dict = response.json()
```

--------------------------------

### Register Local Plugins After Initialization

Source: https://xpublish.readthedocs.io/en/0.4.1/user-guide/plugins

This example demonstrates registering a local plugin after the `xpublish.Rest` application has already been initialized. The `register_plugin()` method is called on the `Rest` instance, allowing for dynamic addition of plugin functionality.

```python
from xpublish import Rest

rest = Rest(datasets)
rest.register_plugin(HelloWorldPlugin())


```

--------------------------------

### Integrate Custom Router with Xpublish and Serve

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-router

This Python code demonstrates how to integrate a custom FastAPI router with an Xpublish dataset and then serve the combined application. It uses `ds.rest(routers=[myrouter])` to incorporate the custom routes and `rest.serve()` to start the server.

```python
import xarray as xr
from fastapi import APIRouter, Depends, HTTPException
from xpublish.dependencies import get_dataset


ds = xr.tutorial.open_dataset(
    'air_temperature',
    chunks={'lat': 5, 'lon': 5},
)

myrouter = APIRouter()


@myrouter.get('/{var_name}/mean')
def get_mean(var_name: str, dataset: xr.Dataset = Depends(get_dataset)):
    if var_name not in dataset.variables:
        raise HTTPException(status_code=404, detail=f"Variable '{var_name}' not found in dataset")

    return float(dataset[var_name].mean())


rest = ds.rest(routers=[myrouter])

rest.serve()

```

--------------------------------

### Load Plugin Routers

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

This method loads application and dataset routers provided by installed plugins. It uses the plugin manager to hook into 'app_router' and 'dataset_router' entry points, returning them as a tuple. Dependencies are passed to the hook calls.

```python
def plugin_routers(self) -> Tuple[List[RouterAndKwargs], List[RouterAndKwargs]]:
    """Load the app and dataset routers for plugins.

    Returns:
        A tuple containing a list of top-level routers from plugins
        and a list of per-dataset routers from plugins
    """
    app_routers = []
    dataset_routers = []

    deps = self.dependencies()

    for router in self.pm.hook.app_router(deps=deps):
        app_routers.append((router, {}))

    for router in self.pm.hook.dataset_router(deps=deps):
        dataset_routers.append((router, {}))

    return app_routers, dataset_routers
```

--------------------------------

### Xpublish Plugin Router Implementation Detail

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-router-plugin

This snippet details the implementation of the `dataset_router` hook within an Xpublish plugin. It constructs a FastAPI router, defines a GET endpoint for calculating variable means, and utilizes dependency injection for the dataset.

```python
from typing import Sequence

from fastapi import APIRouter, Depends, HTTPException

from xpublish import Dependencies, Plugin, SingleDatasetRest, hookimpl


class MeanPlugin(Plugin):
    name: str = 'mean'

    dataset_router_prefix: str = ''
    dataset_router_tags: Sequence[str] = ['mean']

    @hookimpl
    def dataset_router(self, deps: Dependencies):
        router = APIRouter(prefix=self.dataset_router_prefix, tags=list(self.dataset_router_tags))

        @router.get('/{var_name}/mean')
        def get_mean(var_name: str, dataset=Depends(deps.dataset)):
            if var_name not in dataset.variables:
                raise HTTPException(
                    status_code=404,
                    detail=f"Variable '{var_name}' not found in dataset",
                )

            return float(dataset[var_name].mean())


```

--------------------------------

### Get Loaded Plugins (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Returns a dictionary of the currently loaded plugins, mapping plugin names to their respective `Plugin` objects. Dependencies include `Plugin` and `Dict`.

```python
@property
def plugins(self) -> Dict[str, Plugin]:
    """Returns the loaded plugins."""
    return dict(self.pm.list_name_plugin())
```

--------------------------------

### Create App Router Plugin for Plugin Information

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This Python snippet defines an app router plugin 'PluginInfo' that provides top-level API routes. It configures an '/info' prefix and 'info' tags. The 'app_router' hook adds a '/plugins' GET endpoint that returns a dictionary of loaded plugin names and their types.

```python
from fastapi import APIRouter, Depends
from xpublish import Plugin, Dependencies, hookimpl
from typing import Dict


class PluginInfo(Plugin):
    name: str = "plugin_info"

    app_router_prefix = "/info"
    app_router_tags = ["info"]

    @hookimpl
    def app_router(self, deps: Dependencies):
        router = APIRouter(prefix=self.app_router_prefix, tags=self.app_router_tags)

        @router.get("/plugins")
        def plugins(plugins: Dict[str, Plugin] = Depends(deps.plugins)):
            return {name: type(plugin) for name, plugin in plugins.items}

        return router
```

--------------------------------

### Example Router Implementation

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-router-plugin

This snippet shows a basic FastAPI router that calculates the mean of a specified variable within an xarray Dataset. It handles cases where the variable might not exist.

```python
from fastapi import APIRouter, Depends, HTTPException
import xarray as xr

myrouter = APIRouter()

@myrouter.get('/{var_name}/mean')
def get_mean(var_name: str, dataset: xr.Dataset = Depends(get_dataset)):
    if var_name not in dataset.variables:
        raise HTTPException(status_code=404, detail=f"Variable '{var_name}' not found in dataset")

    return float(dataset[var_name].mean())


rest = ds.rest(routers=[myrouter])

```

--------------------------------

### List All Datasets API

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/serving-multiple-datasets

Get a list of all dataset IDs currently being served by the application.

```APIDOC
## GET /datasets

### Description
Returns a list of all dataset IDs (keys) available in the application.

### Method
GET

### Endpoint
/datasets

### Response
#### Success Response (200)
- **dataset_ids** (array) - A list of strings, where each string is a dataset ID.

#### Response Example
{
  "example": "{\"dataset_ids\": [\"air_temperature\", \"rasm\"]}"
}
```

--------------------------------

### Module Version API

Source: https://xpublish.readthedocs.io/en/0.4.1/api/endpoints

Get the versions of key libraries currently loaded in the Xpublish environment.

```APIDOC
## GET /versions

### Description
Returns a dict with currently loaded versions of key libraries.

### Method
GET

### Endpoint
/versions

### Parameters

#### Path Parameters
None

#### Query Parameters
None

### Request Example
None

### Response
#### Success Response (200)
- **versions** (object) - A dictionary mapping library names to their versions.

#### Response Example
```json
{
  "versions": {
    "library1": "1.2.3",
    "library2": "4.5.6"
  }
}
```
```

--------------------------------

### Serve Xarray Dataset with Xpublish (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/index

Demonstrates how to serve an Xarray dataset over HTTP using the xpublish.Rest class or the xarray.Dataset.rest accessor. This snippet shows the server-side setup for publishing datasets.

```python
from xpublish_server import Xpublish
import xarray as xr

dataset = xr.tutorial.load_dataset("air_temperature")
app = Xpublish(datasets={"air_temperature": dataset})

# To run the server:
# uvicorn main:app --reload

```

--------------------------------

### Get Dataset IDs from Plugins and Internal Storage

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Retrieves a list of all dataset IDs, combining those from installed plugins and those stored internally. This function iterates through plugins to collect their dataset IDs and extends the initial list with them.

```python
dataset_ids = list(self._datasets)

for plugin_dataset_ids in self.pm.hook.get_datasets():
    dataset_ids.extend(plugin_dataset_ids)

return dataset_ids
```

--------------------------------

### Initialize and Serve Xarray Dataset with Xpublish

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/introduction

This snippet demonstrates how to import Xpublish, open an Xarray dataset, and then serve it using the `SingleDatasetRest` class or the convenient `xarray.Dataset.rest` accessor. It also shows optional customization of the underlying FastAPI application and server-side cache.

```python
import xarray as xr
import xpublish

ds = xr.tutorial.open_dataset(
    "air_temperature",
    chunks=dict(lat=5, lon=5),
)

# Using SingleDatasetRest class
rest = xpublish.SingleDatasetRest(ds)
rest.serve()

# Or using the xarray.Dataset.rest accessor
ds.rest

ds.rest(
    app_kws=dict(
        title="My Dataset",
        description="Dataset Description",
        openapi_url="/dataset.JSON",
    ),
    cache_kws=dict(available_bytes=1e9),
)
ds.rest.serve()
```

--------------------------------

### Create Xpublish Dataset Provider Plugin with Python

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-provider-plugin

This Python code defines a custom Xpublish plugin that acts as a dataset provider. It implements the `get_datasets` and `get_dataset` hooks to dynamically load datasets from `xr.tutorial` upon request. This approach enhances scalability by avoiding upfront dataset loading. It requires the `xarray` and `xpublish` libraries.

```python
import xarray as xr
from requests import HTTPError

from xpublish import Plugin, Rest, hookimpl


class TutorialDataset(Plugin):
    name: str = 'xarray-tutorial-dataset'

    @hookimpl
    def get_datasets(self):
        return list(xr.tutorial.file_formats)

    @hookimpl
    def get_dataset(self, dataset_id: str):
        try:
            return xr.tutorial.open_dataset(dataset_id)
        except HTTPError:
            return None


rest = Rest({})
rest.register_plugin(TutorialDataset())
rest.serve()

```

--------------------------------

### Initialize FastAPI Application and Routers

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

This function initializes the FastAPI application instance and sets up the application and dataset routers. It includes routers from plugins and applies any configured dataset route prefixes. Dependencies are also initialized.

```python
def _init_app(self) -> FastAPI:
    """Initiate the FastAPI application.

    Returns:
        FastAPI application instance.
    """
    self._app = FastAPI(**self._app_kws)

    self._init_routers(self._routers)
    for rt, kwargs in self._app_routers:
        self._app.include_router(rt, **kwargs)

    self._init_dependencies()

    return self._app
```

--------------------------------

### GET /zarr/.zattrs

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/zarr

Retrieves Zarr attributes for the dataset.

```APIDOC
## GET /.zattrs

### Description
Zarr attributes.

### Method
GET

### Endpoint
`/.zattrs`

### Parameters
#### Query Parameters
None

### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **attributes** (dict) - The Zarr attributes dictionary.

#### Response Example
```json
{
  "dimension_order": ["time", "y", "x"]
}
```
```

--------------------------------

### GET /dataset_info/keys

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/dataset_info

List of the keys (variables) in a dataset.

```APIDOC
## GET /dataset_info/keys

### Description
List of the keys (variables) in a dataset.

### Method
GET

### Endpoint
/dataset_info/keys

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **list[str]** (list) - A list of strings representing the keys in the dataset.

#### Response Example
```json
["temperature", "pressure", "time"]
```
```

--------------------------------

### GET /zarr/.zgroup

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/zarr

Retrieves Zarr group data for the dataset.

```APIDOC
## GET /.zgroup

### Description
Zarr group data.

### Method
GET

### Endpoint
`/.zgroup`

### Parameters
#### Query Parameters
None

### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **metadata** (dict) - The Zarr group metadata.

#### Response Example
```json
{
  "zarr_format": 2,
  "arrays": {},
  "groups": {}
}
```
```

--------------------------------

### GET /zarr/.zmetadata

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/zarr

Retrieves the consolidated Zarr metadata for the dataset.

```APIDOC
## GET /.zmetadata

### Description
Consolidated Zarr metadata.

### Method
GET

### Endpoint
`/.zmetadata`

### Parameters
#### Query Parameters
None

### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **metadata** (dict) - The Zarr metadata dictionary.

#### Response Example
```json
{
  "metadata": {
    "zarr_format": 2,
    "arrays": {},
    "groups": {}
  }
}
```
```

--------------------------------

### Configure Xpublish Plugins (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/manage

This function initializes and configures a given dictionary of plugin classes using provided configuration settings. It takes a dictionary of plugin classes and an optional dictionary of configurations, returning a dictionary of initialized plugin instances. Each plugin is instantiated with keyword arguments from its corresponding configuration.

```python
"""Initialize and configure plugins with given dictionary of configurations."""

from importlib.metadata import entry_points
from typing import Dict, Iterable, Optional, Type

from .hooks import Plugin




[docs]
def find_default_plugins(
    exclude_plugins: Optional[Iterable[str]] = None,
) -> Dict[str, Type[Plugin]]:
    """Find Xpublish plugins from entry point group `xpublish.plugin`.

    Args:
        exclude_plugins: A list of plugin names to ignore.

    Returns:
        A dictionary of plugin names and classes.
    """
    exclude_plugins = set(exclude_plugins or [])

    plugins: Dict[str, Type[Plugin]] = {}

    try:
        plugin_entry_points = entry_points(group='xpublish.plugin')
    except TypeError:
        plugin_entry_points = entry_points()['xpublish.plugin']

    for entry_point in plugin_entry_points:
        if entry_point.name not in exclude_plugins:
            plugins[entry_point.name] = entry_point.load()

    return plugins





[docs]
def load_default_plugins(
    exclude_plugins: Optional[Iterable[str]] = None,
) -> Dict[str, Plugin]:
    """Find and initialize plugins from entry point group `xpublish.plugin`.

    Args:
        exclude_plugins: A list of plugin names to ignore.

    Returns:
        A dictionary of plugin names and instances.
    """
    initialized_plugins: Dict[str, Plugin] = {}

    for name, plugin in find_default_plugins(exclude_plugins=exclude_plugins).items():
        initialized_plugins[name] = plugin()

    return initialized_plugins





[docs]
def configure_plugins(
    plugins: Dict[str, Type[Plugin]],
    plugin_configs: Optional[Dict[str, Dict]] = None,
) -> Dict[str, Plugin]:
    """Initialize and configure plugins with given dictionary of configurations.

    Args:
        plugins: A dictionary of plugin names and classes.
        plugin_configs: A dictionary of plugin names and configurations.

    Returns:
        A dictionary of plugin names and instances.
    """
    initialized_plugins: Dict[str, Plugin] = {}
    plugin_configs = plugin_configs or {}

    for name, plugin in plugins.items():
        kwargs = plugin_configs.get(name, {})
        initialized_plugins[name] = plugin(**kwargs)

    return initialized_plugins

```

--------------------------------

### Initialize FastAPI Application Keyword Arguments (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Sets up the keyword arguments for the FastAPI application. Initializes the application to `None` and an empty dictionary for keyword arguments. If `app_kws` are provided, they update the default settings. Dependencies include `Union` and `dict`.

```python
def init_app_kwargs(self, app_kws: Union[dict, None]) -> None:
    """Set up FastAPI application kwargs.

    Args:
        app_kws: Dictionary of FastAPI application keyword arguments.
    """
    self._app = None
    self._app_kws = {}
    if app_kws is not None:
        self._app_kws.update(app_kws)
```

--------------------------------

### Dataset Info API

Source: https://xpublish.readthedocs.io/en/0.4.1/api/endpoints

Get the dataset schema, similar to the NCO-JSON schema.

```APIDOC
## GET /datasets/{dataset_id}/info

### Description
Dataset schema (close to the NCO-JSON schema).

### Method
GET

### Endpoint
/datasets/{dataset_id}/info

### Parameters

#### Path Parameters
- **dataset_id** (string) - Required - Unique ID of dataset

#### Query Parameters
None

### Request Example
None

### Response
#### Success Response (200)
- **schema** (object) - The dataset schema information.

#### Response Example
```json
{
  "schema": { ... }
}
```
```

--------------------------------

### Dataset HTML Representation API

Source: https://xpublish.readthedocs.io/en/0.4.1/api/endpoints

Get the xarray HTML representation of a dataset.

```APIDOC
## GET /datasets/{dataset_id}/

### Description
Returns the xarray HTML representation of the dataset.

### Method
GET

### Endpoint
/datasets/{dataset_id}/

### Parameters

#### Path Parameters
- **dataset_id** (string) - Required - Unique ID of dataset

#### Query Parameters
None

### Request Example
None

### Response
#### Success Response (200)
- **html_representation** (string) - HTML content representing the dataset.

#### Response Example
```html
<html>...</html>
```
```

--------------------------------

### Xpublish Dataset Router Plugin Structure

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/dataset-router-plugin

This Python snippet demonstrates the basic structure of an Xpublish plugin, inheriting from `xpublish.Plugin` and defining a name. This forms the foundation for creating configurable and loadable plugins.

```python
from xpublish.plugins import Plugin

class DatasetRouterPlugin(Plugin):
    name = "dataset_router"

```

--------------------------------

### Building the Xpublish Dataset Router within a Plugin

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/dataset-router-plugin

This Python snippet demonstrates how to build the dataset router inside an Xpublish plugin. It shows how to use passed arguments for dependencies and initialize the router with a prefix and tags, returning the initialized router.

```python
from xpublish.plugins import Plugin
from pluggy import Hookspec
from starlette.routing import Route
from starlette.responses import JSONResponse
from starlette.exceptions import HTTPException

class DatasetRouterPlugin(Plugin):
    name = "dataset_router"

    prefix: str = "/data"
    tags: list[str] = ["datasets"]

    @Hookspec
    def dataset_router(self, app):
        """Provide a dataset router."""

        async def dataset_info(request):
            dataset_id = request.path_params['dataset_id']
            if dataset_id in request.app.datasets:
                return JSONResponse(content=request.app.datasets[dataset_id].json())
            raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")

        async def datasets_list(request):
            return JSONResponse(content=[d.json() for d in request.app.datasets.values()])

        return [
            Route(f"{self.prefix}/datasets/{{dataset_id}}", endpoint=dataset_info, methods=["GET"]),
            Route(f"{self.prefix}/datasets", endpoint=datasets_list, methods=["GET"]),
        ]

```

--------------------------------

### GET /dataset_info/dict

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/dataset_info

The full dataset as a dictionary, excluding data.

```APIDOC
## GET /dataset_info/dict

### Description
Returns the full dataset as a dictionary, excluding the actual data arrays.

### Method
GET

### Endpoint
/dataset_info/dict

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **dict** (dict) - A dictionary representation of the dataset's metadata.

#### Response Example
```json
{
  "coords": {
    "time": {
      "dims": ["time"],
      "attrs": {},
      "dtype": "int64"
    }
  },
  "dims": {
    "time": 10
  },
  "variables": {
    "temperature": {
      "dims": ["time"],
      "attrs": {},
      "dtype": "float64"
    },
    "pressure": {
      "dims": ["time"],
      "attrs": {},
      "dtype": "float64"
    }
  },
  "attrs": {
    "description": "Sample dataset"
  }
}
```
```

--------------------------------

### Customize FastAPI App and Cache for Xpublish

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/introduction

Illustrates optional customization of the underlying FastAPI application and the server-side cache when using the `.rest` accessor.

```python
ds.rest(
    app_kws=dict(
        title="My Dataset",
        description="Dataset Description",
        openapi_url="/dataset.JSON",
    ),
    cache_kws=dict(available_bytes=1e9),
)
```

--------------------------------

### GET /dataset_info/

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/dataset_info

Returns the xarray HTML representation of the dataset.

```APIDOC
## GET /dataset_info/

### Description
Returns the xarray HTML representation of the dataset.

### Method
GET

### Endpoint
/dataset_info/

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **HTMLResponse** (HTMLResponse) - The HTML representation of the dataset.

#### Response Example
None
```

--------------------------------

### Load and Initialize Xpublish Plugins (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/manage

This function first finds all available Xpublish plugins using `find_default_plugins` and then initializes each discovered plugin. It returns a dictionary containing instances of the loaded plugins, keyed by their names. Exclusions specified in `exclude_plugins` are passed down to the `find_default_plugins` function.

```python
"""Find and initialize plugins from entry point group `xpublish.plugin`."""

from importlib.metadata import entry_points
from typing import Dict, Iterable, Optional, Type

from .hooks import Plugin




[docs]
def find_default_plugins(
    exclude_plugins: Optional[Iterable[str]] = None,
) -> Dict[str, Type[Plugin]]:
    """Find Xpublish plugins from entry point group `xpublish.plugin`.

    Args:
        exclude_plugins: A list of plugin names to ignore.

    Returns:
        A dictionary of plugin names and classes.
    """
    exclude_plugins = set(exclude_plugins or [])

    plugins: Dict[str, Type[Plugin]] = {}

    try:
        plugin_entry_points = entry_points(group='xpublish.plugin')
    except TypeError:
        plugin_entry_points = entry_points()['xpublish.plugin']

    for entry_point in plugin_entry_points:
        if entry_point.name not in exclude_plugins:
            plugins[entry_point.name] = entry_point.load()

    return plugins





[docs]
def load_default_plugins(
    exclude_plugins: Optional[Iterable[str]] = None,
) -> Dict[str, Plugin]:
    """Find and initialize plugins from entry point group `xpublish.plugin`.

    Args:
        exclude_plugins: A list of plugin names to ignore.

    Returns:
        A dictionary of plugin names and instances.
    """
    initialized_plugins: Dict[str, Plugin] = {}

    for name, plugin in find_default_plugins(exclude_plugins=exclude_plugins).items():
        initialized_plugins[name] = plugin()

    return initialized_plugins

```

--------------------------------

### GET /dataset_info/info

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/dataset_info

Dataset schema information, similar to the NCO-JSON schema.

```APIDOC
## GET /dataset_info/info

### Description
Provides dataset schema information, closely resembling the NCO-JSON schema. Includes dimensions, variables with their types, dimensions, and attributes, as well as global attributes.

### Method
GET

### Endpoint
/dataset_info/info

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **dict** (dict) - A dictionary containing the dataset's schema information.

#### Response Example
```json
{
  "dimensions": {
    "time": 10
  },
  "variables": {
    "temperature": {
      "type": "float64",
      "dimensions": ["time"],
      "attributes": {}
    },
    "pressure": {
      "type": "float64",
      "dimensions": ["time"],
      "attributes": {}
    }
  },
  "global_attributes": {
    "description": "Sample dataset"
  }
}
```
```

--------------------------------

### Initialize Cache Keyword Arguments (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Sets up the keyword arguments for the cache. Initializes the cache to `None` and sets default `available_bytes` to 1MB. If `cache_kws` are provided, they update the default settings. Dependencies include `Union` and `dict`.

```python
def init_cache_kwargs(self, cache_kws: Union[dict, None]) -> None:
    """Set up cache kwargs.

    Args:
        cache_kws: Dictionary of cache keyword arguments.
    """
    self._cache = None
    self._cache_kws = {'available_bytes': 1e6}
    if cache_kws is not None:
        self._cache_kws.update(cache_kws)
```

--------------------------------

### GET /zarr/{var}/{chunk}

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/plugins/included/zarr

Retrieves a specific Zarr array chunk for a given variable. Supports caching.

```APIDOC
## GET /{var}/{chunk}

### Description
Get a zarr array chunk.

This will return cached responses when available.

### Method
GET

### Endpoint
`/{var}/{chunk}`

### Parameters
#### Path Parameters
- **var** (str) - Required - Variable in dataset
- **chunk** (str) - Required - Zarr chunk identifier

#### Query Parameters
None

### Request Body
None

### Request Example
None

### Response
#### Success Response (200)
- **data** (bytes) - The chunk data as bytes.

#### Response Example
```
(binary chunk data)
```

#### Error Response (404)
- **detail** (str) - 'No subgroups' if attempting to access a subgroup via this endpoint.
```

--------------------------------

### Create App Router Plugin in Python

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

This Python code defines a plugin 'PluginInfoPlugin' to add top-level routes to the xpublish server. It introduces a '/info/plugins' endpoint that returns a dictionary of registered plugin names and their corresponding class names. Dependencies include 'fastapi' and 'xpublish'. It takes a dictionary of plugins as input and returns a dictionary of plugin information.

```python
from fastapi import APIRouter, Depends
from typing import Dict
from xpublish import Plugin, Dependencies, hookimpl


class PluginInfoPlugin(Plugin):
    name: str = "plugin_info"

    app_router_prefix: str = "/info"
    app_router_tags: list = ["info"]

    @hookimpl
    def app_router(self, deps: Dependencies):
        router = APIRouter(
            prefix=self.app_router_prefix,
            tags=self.app_router_tags
        )

        @router.get("/plugins")
        def get_plugins(plugins: Dict[str, Plugin] = Depends(deps.plugins)):
            return {name: type(plugin).__name__ for name, plugin in plugins.items()}

        return router


# Use with Rest instance
from xpublish import Rest
rest = Rest({"dataset1": ds})
rest.register_plugin(PluginInfoPlugin())
rest.serve()

# Access app-level endpoint
# curl http://0.0.0.0:9000/info/plugins
```

--------------------------------

### Serve Multiple Datasets with Xpublish and FastAPI

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/serving-multiple-datasets

Demonstrates how to initialize and serve multiple xarray Datasets using xpublish.Rest within a FastAPI application. Requires the xarray and xpublish libraries.

```python
import xarray as xr
import xpublish

ds = xr.tutorial.open_dataset("air_temperature")
ds2 = xr.tutorial.open_dataset("rasm")

rest_collection = xpublish.Rest({"air_temperature": ds, "rasm": ds2})

rest_collection.serve()
```

--------------------------------

### Existing Dataset Router

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/dataset-router-plugin

This snippet shows the initial structure of a dataset router before it is converted into a plugin. It defines the core logic for routing datasets.

```python
    @app.route("/datasets/<dataset_id>")
    async def dataset_info(request, dataset_id):
        """Return information about a specific dataset."""
        if dataset_id in request.app.datasets:
            return JSONResponse(content=request.app.datasets[dataset_id].json())
        raise HTTPException(status_code=404, detail=f"Dataset {dataset_id} not found")

    @app.route("/datasets")
    async def datasets_list(request):
        """Return a list of available datasets."""
        return JSONResponse(content=[d.json() for d in request.app.datasets.values()])

```

--------------------------------

### Import Xpublish and Open Xarray Dataset

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/introduction

Imports the necessary libraries (xarray and xpublish) and opens a sample Xarray dataset with specified chunking for server-side operations.

```python
import xarray as xr
import xpublish

ds = xr.tutorial.open_dataset(
    "air_temperature",
    chunks=dict(lat=5, lon=5),
)
```

--------------------------------

### Get Cache Instance (Python)

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Returns the `cachey.Cache` instance used by the FastAPI application. If the cache has not been initialized, it creates a new instance using the configured keyword arguments. Dependencies include `cachey.Cache`.

```python
@property
def cache(self) -> cachey.Cache:
    """Returns the :class:`cachey.Cache` instance used by the FastAPI application."""
    if self._cache is None:
        self._cache = cachey.Cache(**self._cache_kws)
    return self._cache
```

--------------------------------

### Load Xpublish Plugin Directly in Initializer

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This snippet demonstrates how to load a plugin directly when initializing the xpublish.Rest class. It assumes 'datasets' is defined and 'HelloWorldPlugin' is an available plugin class.

```python
from xpublish import Rest

rest = Rest(datasets, plugins={"hello-world": HelloWorldPlugin()})
```

--------------------------------

### Register Xpublish Plugin Dynamically

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This snippet shows how to initialize xpublish.Rest without plugins and then register a plugin afterwards using the 'register_plugin' method. It assumes 'datasets' is defined and 'HelloWorldPlugin' is an available plugin class.

```python
from xpublish import Rest

rest = Rest(datasets)
rest.register_plugin(HelloWorldPlugin())
```

--------------------------------

### Serve Multiple Datasets with Custom Routers

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/serving-multiple-datasets

Shows how to serve multiple datasets using Xpublish's Rest class while also incorporating custom routers. This demonstrates that custom route handling remains consistent whether serving one or multiple datasets.

```python
import xarray as xr
import xpublish

# Assuming 'myrouter' is a pre-defined FastAPI router
# from fastapi import APIRouter
# myrouter = APIRouter()

ds = xr.tutorial.open_dataset("air_temperature")
ds2 = xr.tutorial.open_dataset("rasm")

rest_collection = xpublish.Rest(
    {"air_temperature": ds, "rasm": ds2},
    routers=[myrouter]  # Example: Pass custom routers here
)

rest_collection.serve()
```

--------------------------------

### Publish Xarray Dataset with SingleDatasetRest

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/introduction

Demonstrates how to publish an Xarray Dataset using the `xpublish.SingleDatasetRest` class for server-side data serving.

```python
rest = xpublish.SingleDatasetRest(ds)
```

--------------------------------

### Create Dataset Provider Plugin in Python

Source: https://context7.com/context7/xpublish_readthedocs_io_en_0_4_1/llms.txt

This Python code defines a 'TutorialDatasetProvider' plugin that allows xpublish to dynamically load datasets on demand. It implements `get_datasets` to list available dataset IDs and `get_dataset` to open a specific dataset using `xarray.tutorial.open_dataset`. Dependencies include 'xarray' and 'xpublish'. It returns a list of dataset IDs or a dataset object.

```python
import xarray as xr
from requests import HTTPError
from xpublish import Plugin, Rest, hookimpl
from xpublish.utils.api import DATASET_ID_ATTR_KEY


class TutorialDatasetProvider(Plugin):
    name: str = 'xarray-tutorial-dataset'

    @hookimpl
    def get_datasets(self):
        """Return list of available dataset IDs"""
        return list(xr.tutorial.file_formats)

    @hookimpl
    def get_dataset(self, dataset_id: str):
        """Return dataset for given ID, or None if not found"""
        try:
            ds = xr.tutorial.open_dataset(dataset_id)
            # Set unique ID attribute for caching
            return ds.assign_attrs({
                DATASET_ID_ATTR_KEY: f"{self.name}_{dataset_id}"
            })
        except HTTPError:
            return None


# Create server with dataset provider
rest = Rest({})
rest.register_plugin(TutorialDatasetProvider())
rest.serve()

# Datasets are loaded on first access
# curl http://0.0.0.0:9000/datasets
# curl http://0.0.0.0:9000/datasets/air_temperature/info
```

--------------------------------

### Publishing a Single xarray Dataset with xpublish.SingleDatasetRest

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/api/index

The `xpublish.SingleDatasetRest` class is a specialized version of `xpublish.Rest` designed for serving only a single xarray.Dataset. It simplifies the process when you have one dataset to publish. Similar to `Rest`, it offers methods for setup and customization.

```python
import xarray as xr
import xpublish

# Create or load an xarray Dataset
ds = xr.Dataset()

# Initialize the SingleDatasetRest class
# single_dataset_publisher = xpublish.SingleDatasetRest(datasets=ds)

# Access methods and properties
# app = single_dataset_publisher.app
# cache_config = single_dataset_publisher.cache
# single_dataset_publisher.serve()

```

--------------------------------

### Xpublish Plugin Subclassing

Source: https://xpublish.readthedocs.io/en/0.4.1/getting-started/tutorial/dataset-router-plugin

This snippet demonstrates the basic structure for creating an Xpublish plugin by subclassing `xpublish.Plugin` and assigning a unique name. This serves as the foundation for defining plugin behavior and configuration.

```python
from fastapi import APIRouter, Depends, HTTPException

from xpublish import Dependencies, Plugin, SingleDatasetRest, hookimpl


class MeanPlugin(Plugin):
    name: str = 'mean'


```

--------------------------------

### Registering an Xpublish Plugin Directly

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/getting-started/tutorial/dataset-router-plugin

This Python snippet shows how to directly register an Xpublish plugin. This method is suitable for plugins that are not intended for distribution and allows them to be loaded by the Xpublish application.

```python
from xpublish_server import XpublishServer
from .dataset_router_plugin import DatasetRouterPlugin

app = XpublishServer(plugins=[DatasetRouterPlugin()])

```

--------------------------------

### Get FastAPI Dependencies for Plugins

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

This method creates and returns a Dependencies object, which encapsulates the necessary FastAPI dependencies to be passed to plugin router methods. It provides access to dataset IDs, the dataset itself, cache, plugins, and the plugin manager.

```python
def dependencies(self) -> Dependencies:
    """FastAPI dependencies to pass to plugin router methods.

    Returns:
        initialized :class:xpublish.plugins.Dependencies object.
    """
    deps = Dependencies(
        dataset_ids=self.get_datasets_from_plugins,
        dataset=self._get_dataset_func,
        cache=lambda: self.cache,
        plugins=lambda: self.plugins,
        plugin_manager=lambda: self.pm,
    )

    return deps
```

--------------------------------

### Python: Initialize FastAPI Application with Cache and Plugins

Source: https://xpublish.readthedocs.io/en/0.4.1/_modules/xpublish/rest

Initializes a FastAPI application with support for caching and plugins. It accepts dictionaries for cache and application keyword arguments, and an optional dictionary for pre-configured plugins. If no plugins are desired, an empty dictionary should be provided. The cache defaults to 1MB but can be configured via 'available_bytes'.

```python
def __init__(
    self,
    dataset: Dataset,
    routers: List[APIRouter] = None,
    cache_kws: dict = None,
    app_kws: dict = None,
    plugins: dict = None,
):
        """
        Initialize the Xpublish dataset application.

        Parameters
        ----------
        dataset
            The dataset to serve.
        routers
            List of FastAPI routers to include.
        cache_kws: Dictionary of keyword arguments to be passed to
            :meth:`cachey.Cache.__init__()`. By default, the cache size is set
            to 1MB, but this can be changed with ``available_bytes``.
        app_kws: Dictionary of keyword arguments to be passed to
            :meth:`fastapi.FastAPI.__init__()`.
        plugins: Optional dictionary of loaded, configured plugins. Overrides
            automatic loading of plugins. If no plugins are desired, set to an
            empty dict.
        """
        self._dataset = dataset

        super().__init__({}, routers, cache_kws, app_kws, plugins)
```

--------------------------------

### Implement Dataset Info Plugin with Prefix and Tags

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This Python snippet defines a dataset router plugin 'DatasetInfoPlugin' that adds dataset-specific information routes. It configures a '/info' prefix and 'info' tags for its routes. The 'dataset_router' hook implements a '/keys' endpoint to list dataset variables.

```python
from fastapi import APIRouter, Depends
from xpublish import Plugin, Dependencies, hookimpl


class DatasetInfoPlugin(Plugin):
    name: str = "dataset-info"

    dataset_router_prefix = "/info"
    dataset_router_tags = ["info"]

    @hookimpl
    def dataset_router(self, deps: Dependencies):
        router = APIRouter(
            prefix=self.dataset_router_prefix, tags=self.dataset_router_tags
        )

        @router.get("/keys")
        def list_keys(dataset=Depends(deps.dataset)):
            return dataset.variables

        return router
```

--------------------------------

### Create Dataset Router Plugin with Custom Attributes

Source: https://xpublish.readthedocs.io/en/0.4.1/_sources/user-guide/plugins

This Python snippet defines a dataset router plugin named 'dataset-attrs'. It uses the 'dataset_router' hook to add a '/attrs' GET endpoint that returns the attributes of a dataset. It demonstrates dependency injection for accessing the dataset and customizes router prefix and tags.

```python
from fastapi import APIRouter, Depends
from xpublish import Plugin, Dependencies, hookimpl


class DatasetAttrs(Plugin):
    name: str = "dataset-attrs"

    @hookimpl
    def dataset_router(self, deps: Dependencies):
        router = APIRouter()

        @router.get("/attrs")
        def get_attrs(ds=Depends(deps.dataset)):
            return ds.attrs

        return router
```
Tokens:
Raw