Is possible to modify a dataset once the serve is started?

[josephnowak]

Hi, I'm interested in use xpublish to analyze some financial data that I have in a cluster. The thing is that the data is a time-series, so every day the data suffer a concatenation of data and when this happens the dataset that I made public does not show the new dates. I want to know how can I update that dataset without killing the serve and public again the datasets. I really can't find a method to do this in the documentation, sorry for annoying you with this but I think that this a great API for my use case.

[josephnowak]

I did a code that does exactly what I wanted, but It makes use of private attributes so I suppose It is not a good idea to use it and probably could produce some other problems, but It exemplifies what I want.

import numpy as np 
import xarray
import time
import xpublish
import zarr
import threading
import os
from fsspec.implementations.http import HTTPFileSystem


def compare_dataset(a, b):
	b = b.to_array()
	a = a.to_array()
	equals = True
	for name, coord in a.coords.items():
		equals &= coord.equals(b.coords[name])
	equals &= np.allclose(a.values, b.values, equal_nan=True)
	return equals



def generate_zarr(file_name, shape):
	a = np.random.rand(*shape)
	b = xarray.DataArray(
	    a,
	    dims=['index', 'columns'],
	    coords={'index': list(range(shape[0])), 'columns': list(range(shape[1]))}
	).to_dataset(name='test_arr').chunk({'index': 5})
	b.to_zarr(file_name, consolidated=True, mode='w')


generate_zarr('testing.zarr', (10, 5))
generate_zarr('testing2.zarr', (15, 7))
generate_zarr('testing3.zarr', (4, 7))

d = xarray.open_zarr('testing.zarr', consolidated=True)
d2 = xarray.open_zarr('testing2.zarr', consolidated=True)
d3 = xarray.open_zarr('testing3.zarr', consolidated=True)


# d.rest.serve(host='127.0.0.1', port=9000)
rest_collection = xpublish.Rest({'test': d})

# rest_collection.serve(host='127.0.0.1', port=9000)
p = threading.Thread(target=rest_collection.serve, kwargs={'host': '127.0.0.1', 'port': 9000})
p.start()

fs = HTTPFileSystem()

while True:
	key = int(input())
	if key == 1:	
		# Putting the original dataset
		d.attrs['_xpublish_id'] = 'test'
		rest_collection._datasets['test'] = d 

		# validating the results
		http_map = fs.get_mapper('http://127.0.0.1:9000/datasets/test')
		test_d = xarray.open_zarr(http_map, consolidated=True)
		print(compare_dataset(test_d, d))
	elif key == 2:
		d2.attrs['_xpublish_id'] = 'test'
		rest_collection._datasets['test'] = d2 

		# validating the results
		http_map = fs.get_mapper('http://127.0.0.1:9000/datasets/test')
		test_d = xarray.open_zarr(http_map, consolidated=True)
		print(compare_dataset(test_d, d2))
	elif key == 3:
		d3.attrs['_xpublish_id'] = 'test_evo'
		rest_collection._datasets['test_evo'] = d3 

		# validating the results
		http_map = fs.get_mapper('http://127.0.0.1:9000/datasets/test_evo')
		test_d = xarray.open_zarr(http_map, consolidated=True)
		print(compare_dataset(test_d, d3))
	elif key == 4:
		# Kill all
		os._exit(-1)

	# showing the datasets
	print(rest_collection._datasets)
	rest_collection.cache.clear()

[jhamman]

I haven't done anything like this yet. I wonder if either @lsetiawan or @benbovy has?

[benbovy]

@josephnowak you could update the dataset being served using a custom API endpoint, like in the example below where /add-random-var adds a new variable to the dataset and returns the variable name:

import string
import random

import numpy as np
import xarray as xr
import xpublish
from xpublish.dependencies import get_dataset
from xpublish.routers import base_router
from fastapi import APIRouter, Depends


update_router = APIRouter()


def create_random_var():
    vname = ''.join(random.choice(string.ascii_lowercase) for i in range(7))
    var = ('x', np.random.rand(100))
    return vname, var


@update_router.post("/add-random-var")
def add_random_var(dataset: xr.Dataset = Depends(get_dataset)) -> str:
    vname, var = create_random_var()
    dataset[vname] = var
    return vname
    

ds = xr.Dataset(dict([create_random_var()]))

ds.rest(routers=[base_router, update_router])

ds.rest.serve(host="127.0.0.1")

You could then have a service or a process that sends a request to this API endpoint every day.

I guess you could use some other features in FastAPI (authentication, origin, etc.) to restrict the access to this endpoint.

[benbovy]

The current limitation is that you can not replace the whole dataset being served by another one. All updates have to be in-place.

Also, it's not possible yet to use the application to update a collection of datasets being served, but I think it would be straightforward to support it by adding a get_dataset_collection FastAPI dependency to Xpublish.