diff --git a/Makefile b/Makefile index fe8cafc3..1411d4f6 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: install assets lite-build lite-dev lite-test bro pub-dir pub-color pub-gris sub +.PHONY: install assets lite-build lite-dev lite-test run VENV = ./venv BIN = $(VENV)/bin @@ -38,18 +38,6 @@ lite-test: $(MAKE) lite-build -# To run the different services, for convenience -bro: - ${BIN}/python3 -m caterva2.services.bro - -pub-dir: - ${BIN}/python3 -m caterva2.services.pub --id dir - -pub-color: - ${BIN}/python3 -m caterva2.services.pub --id color - -pub-gris: - ${BIN}/python3 -m caterva2.services.pub --id gris - -sub: +# To run the server, for convenience +run: BLOSC_TRACE=1 ${BIN}/python3 -m caterva2.services.sub diff --git a/README-DEVELOPERS.md b/README-DEVELOPERS.md index 2e410a9d..5a6e61c8 100644 --- a/README-DEVELOPERS.md +++ b/README-DEVELOPERS.md @@ -12,7 +12,7 @@ As the config files are already there, this essentially boils down to: pre-commit install ``` -Also, for running the tests, one needs to run manually the broker, publisher and subscriber. +Also, for running the tests, one needs to run manually the subscriber. There is a `caterva2.tests.services` script that does this. ## Build CSS and JS @@ -67,33 +67,6 @@ USE_REQUESTS=1 CATERVA2_SECRET=c2sikrit pytest ``` ``` -### With external daemons - -To have daemons running across several test runs (for faster testing), start the daemons: - -```shell -python -m caterva2.tests.services & -``` - -or, if you prefer: - -```shell -cat2bro & -cat2pub foo root-example & -cat2sub & -``` - -State files will be stored in dir `_caterva2/`. - -Finally, in another shell (unless you like to hear the daemons chatting), run the tests: - -```shell -env CATERVA2_USE_EXTERNAL=1 python -m pytest -s -``` - -For stopping the daemons, you will have to kill the `caterva2.tests.services` process. -If you started them manually, you will have to kill them manually too (sorry!). - ## Build wheels We are using [hatch](https://hatch.pypa.io) as the build system, so for building wheels and @@ -118,12 +91,7 @@ cd .. # to avoid using the source code python -m caterva2.tests -v ``` -Please note that the services should be not running at this point. In case you want to check against -the current services, you can do: - -```shell -env CATERVA2_USE_EXTERNAL=1 python -m caterva2.tests -v -``` +Please note that the services should be not running at this point. ## Create docs diff --git a/README-PUBSUB.md b/README-PUBSUB.md deleted file mode 100644 index c07aa838..00000000 --- a/README-PUBSUB.md +++ /dev/null @@ -1,148 +0,0 @@ -# The Publisher-Subscriber Model in Caterva2 -**Note**: this functionality is in alpha stage; if you are interested in testing it, please contact the ironArray team. - -Caterva2 offers the possibility of using the publisher-subscriber (pub-sub) message pattern. Under this framework, when a user uses a client (Web API, Python API or command line) to query datasets, the client will connect to a Caterva2 **subscriber** service, which in turn will communicate with the associated **publishers** to which it is subscribed, to retrieve the requested datasets. This subscriber/publisher interaction is mediated by a **broker** service. - -In order to set up a Caterva2 deployment to enable the publisher-subscriber model on your system, you will thus need the following components: - -- One **broker** service to enable the communication between publishers and subscribers. -- Several **publishers**, each one providing subscribers with access to one root and the datasets that it contains. The root may be a native Caterva2 directory with Blosc2 and plain files, or an HDF5 file (support for other formats may be added). -- Several **subscribers**, each one tracking changes in multiple roots and datasets from publishers, and caching them locally for efficient reuse. -- Several **clients**, each one asking a subscriber to track roots and datasets, and provide access to their data and metadata. - -Publishers and subscribers may be far apart, in different networks with limited or expensive connectivity between them, while subscribers and clients will usually be close enough to have fast and cheap connectivity (e.g. a local network). - -## Installation -To support this additional functionality, it is necessary to install Caterva2 with the `[services,clients]` extra features added to the last argument of `pip` commands detailed in the [README](https://github.com/ironArray/Caterva2?tab=readme-ov-file#installation). There are also additional options which are of interest if working under the pub-sub model. - -- `hdf5` to enable serving HDF5 files as Caterva2 roots at the publisher -- `services` for running all Caterva2 services (broker, publisher, subscriber) -- `base-services` for running the Caterva2 broker or publisher services (lighter, less dependencies) - -## Quick start - -(Find more detailed step-by-step [tutorials](Tutorials) in Caterva2 documentation.) - -For the purpose of this quick start, let's use the datasets within the `root-example` folder: - -```sh -cd Caterva2 -ls -F root-example/ -``` - -``` -README.md dir2/ ds-1d-fields.b2nd ds-2d-fields.b2nd ds-sc-attr.b2nd -dir1/ ds-1d-b.b2nd ds-1d.b2nd ds-hello.b2frame -``` - -Now: - -- create a virtual environment and install Caterva2 with the `[services,clients]` extras (see above). -- copy the configuration file `caterva2.sample.toml` to `caterva2.toml` and edit to your - needs (see the fully documented `caterva2.sample.toml` file and [caterva2.toml](caterva2.toml) for help). - -Then fire up the broker, start publishing a root named `foo` with `root-example` datasets, and create a subscriber: - -```sh -cat2bro & # broker -cat2pub foo root-example & # publisher -CATERVA2_SECRET=c2sikrit cat2sub & # subscriber -``` -(To stop them later on, bring each one to the foreground with `fg` and press Ctrl+C.) - -To create a user, you can use the `cat2adduser` command line client. For example: - -```sh -cat2adduser user@example.com foobar11 -``` - -We can then examine a file in the `foo` root, which is being published by the publisher: - -```sh -cat2cli --user "user@example.com" --pass "foobar11" info foo/README.md -``` - -### Pub-sub in the command line client -Now that the services are running, we can use the `cat2cli` client to talk -to the subscriber. In another shell, let's list all the available roots in the system: - -```sh -cat2cli roots -``` - -``` -foo -``` - -We only have the `foo` root that we started publishing. If other publishers were running, -we would see them listed here too. - -Let's ask our local subscriber to subscribe to the `foo` root: - -```sh -cat2cli --username user@example.com --password foobar11 subscribe foo # -> Ok -``` - -Now, one can list the datasets in the `foo` root: - -```sh -cat2cli --username user@example.com --password foobar11 list foo -``` - -``` -kevlar.h5 -kevlar/!_attrs_.json -kevlar/entry/!_attrs_.json -kevlar/entry/data/!_attrs_.json -kevlar/entry/data/data.b2nd -``` - -Let's ask the subscriber for more info about the `foo/dir2/ds-4d.b2nd` dataset: - -```sh -cat2cli --username user@example.com --password foobar11 info @shared/kevlar/entry/data/data.b2nd -``` - -``` -{ - 'shape': [1000, 2167, 2070], - 'chunks': [1, 2167, 2070], - 'blocks': [1, 11, 2070], - 'dtype': 'uint16', - 'schunk': { - 'cbytes': 0, - 'chunkshape': 4485690, - 'chunksize': 8971380, - 'contiguous': True, - 'cparams': { - 'codec': 5, - 'codec_meta': 0, - 'clevel': 1, - 'filters': [0, 0, 0, 0, 0, 1], - 'filters_meta': [0, 0, 0, 0, 0, 0], - 'typesize': 2, - 'blocksize': 45540, - 'nthreads': 1, - 'splitmode': 3, - 'tuner': 0, - 'use_dict': False, - 'filters, meta': [[1, 0]] - }, - 'cratio': 0.0, - 'nbytes': 8971380000, - 'urlpath': '/Users/faltet/blosc/Caterva2/_caterva2/sub/shared/kevlar/entry/data/data.b2nd', - 'vlmeta': {'_ftype': 'hdf5', '_dsetname': 'entry/data/data'}, - 'nchunks': 1000, - 'mtime': None - }, - 'mtime': '2025-05-27T11:33:12.287605Z' -} -``` - -This command returns a JSON object with the dataset's metadata, including its shape, chunks, blocks, data type, and compression parameters. The `schunk` field contains information about the underlying Blosc2 super-chunk that stores the dataset's data. - -There are more commands available in the `cat2cli` client; ask for help with: - -```sh -cat2cli --help -``` diff --git a/README.md b/README.md index 1c0710e3..4802b87f 100644 --- a/README.md +++ b/README.md @@ -96,14 +96,6 @@ python -m pytest -v Tests will use a copy of Caterva2's `root-example` directory. After they finish, state files will be left under the `_caterva2_tests` directory for inspection (it will be re-created when tests are run again). -In case you want to run the tests with your own running daemons, you can do: - -```shell -env CATERVA2_USE_EXTERNAL=1 python -m caterva2.tests -v -``` - -Neither `root-example` nor `_caterva2_tests` will be used in this case. - ## Quick start (Find more detailed step-by-step [tutorials](Tutorials) in Caterva2 documentation.) @@ -123,9 +115,10 @@ dir1/ ds-1d-b.b2nd ds-1d.b2nd ds-hello Now: - create a virtual environment and install Caterva2 with the `[subscriber,clients]` extras (see above). -- copy the configuration file `caterva2-standalone.sample.toml` to `caterva2.toml`. +- copy the configuration file `caterva2.sample.toml` to `caterva2.toml`. -For more advanced configuration options, see the fully documented `caterva2.sample.toml` file (see also [caterva2.toml](caterva2.toml) in Caterva2 tutorials). Subscribers (and clients, to a limited extent) may get their configuration from a `caterva2.toml` file at the current directory (or an alternative file given with the `--conf` option). +Subscribers (and clients, to a limited extent) may get their configuration from a `caterva2.toml` file at the current directory (or an alternative file given with the `--conf` option). +See also [configuration.md](configuration.md) in Caterva2 tutorials. Then run the subscriber: diff --git a/RELEASING.rst b/RELEASING.rst index 43541316..7020c20e 100644 --- a/RELEASING.rst +++ b/RELEASING.rst @@ -31,7 +31,7 @@ installing the wheel, then test it:: Check that the examples in docstrings are up to date. You will need to register a user in https://cat2.cloud/demo/ with username 'user@example.com' and password 'foo'. Then, copy -the content of ``caterva2-standalone.sample.toml`` to ``caterva2.toml`` +the content of ``caterva2.sample.toml`` to ``caterva2.toml`` and run the following commands:: $ rm -r _caterva2/ diff --git a/caterva2-standalone.sample.toml b/caterva2-standalone.sample.toml deleted file mode 100644 index 84567997..00000000 --- a/caterva2-standalone.sample.toml +++ /dev/null @@ -1,29 +0,0 @@ -# Example configuration for a standalone subscriber -# -# It's possible to run only the subscriber. Then the configuration has only a -# section for the subscriber. And maybe another one for the client. - -# The subscriber section must define: -# -# - statedir: the directory where the subcriber's data will be stored (default: _caterva2/sub) -# - http: where the subscriber listens to (a unix socket or a host/port) (default: localhost:8002) -# - urlbase: the base url users will use to reach the subscriber (default: http://localhost:8002) -# - quota: if defined, it will limit the disk usage (default: 0, no limit) -# - maxusers: if defined, it will limit the number of users (default: 0, no limit) -# - login: if true, users will need to authenticate (default: true) -# - register: if true, users will be able to register (default: false) -# -[subscriber] -statedir = "_caterva2/sub" -#http = "_caterva2/sub/uvicorn.socket" -http = "localhost:8002" -urlbase = "http://localhost:8002" -quota = "10G" -maxusers = 5 -register = true # allow users to register - -# The client section defines the credentials for the client to authenticate -# against the subscriber. -[client] -username = "" -password = "" diff --git a/caterva2.sample.toml b/caterva2.sample.toml index 1c552cee..84567997 100644 --- a/caterva2.sample.toml +++ b/caterva2.sample.toml @@ -1,54 +1,29 @@ -# Example configuration file for Caterva2 components. +# Example configuration for a standalone subscriber # -# This may be parsed by different programs, and each program may look up settings in its own section, or in other programs' sections, if present. For instance, there is no setting in the ``subscriber`` section for the broker endpoint; instead, the subscriber program will look ``broker.http`` up. For instance, in a subscriber configuration file:: -# -# [broker] -# http = ... # Broker HTTP endpoint, to be used by subscriber. -# # No need for more broker settings unless the broker is to use this file. -# -# [subscriber] -# ... +# It's possible to run only the subscriber. Then the configuration has only a +# section for the subscriber. And maybe another one for the client. + +# The subscriber section must define: # -# Some sections may appear multiple times, each with a different ID (see below). However, if you are to use a single program of each category, you should be file with ID-less sections. +# - statedir: the directory where the subcriber's data will be stored (default: _caterva2/sub) +# - http: where the subscriber listens to (a unix socket or a host/port) (default: localhost:8002) +# - urlbase: the base url users will use to reach the subscriber (default: http://localhost:8002) +# - quota: if defined, it will limit the disk usage (default: 0, no limit) +# - maxusers: if defined, it will limit the number of users (default: 0, no limit) +# - login: if true, users will need to authenticate (default: true) +# - register: if true, users will be able to register (default: false) # -# All sections and settings are optional. - -# The configuration of the broker service. -# Only one of these is allowed for the moment. -[broker] -http = "localhost:8000" # The ``host:port`` endpoint where the service listens for HTTP requests. Other programs may look ``broker.http`` up to find how to connect to a broker. Use ``*`` as a host to listen on all addresses. -statedir = "_caterva2/bro" # The directory where the service will place state files. -loglevel = "warning" # All service messages having this severity or worse will be logged. - -# The configuration of the publisher service. -# Several of these are allowed, each with a different ID (the string after the dot). A publisher invoked with ``--id=something`` will look its configuration up in the ``publisher.something`` section. -[publisher.1] -http = "localhost:8001" # The ``host:port`` endpoint where the service listens for HTTP requests. Use ``*`` as a host to listen on all addresses. -statedir = "_caterva2/pub" # The directory where the service will place state files. -loglevel = "warning" # All service messages having this severity or worse will be logged. -name = "foo" # The name given to the root to be registered at the broker. This setting has no default, if it is not defined here, you need to give it to the publisher as an argument. -root = "root-examples" # The location (directory, HDF5 file...) containing the datasets for the registered root. - -# Only one of these is allowed. It will be used by a publisher invoked with no ID, or it may be used by other programs to find how to connect to a publisher (``publisher.http``). -[publisher] -http = "localhost:8001" -# ... other settings as above ... - -# The configuration of the subscriber service. -# Several of these are allowed, each with a different ID (the string after the dot). A subscriber invoked with ``--id=something`` will look its configuration up in the ``subscriber.something`` section. -[subscriber.1] -http = "localhost:8002" # The ``host:port`` endpoint where the service listens for HTTP requests. Use ``*`` as a host to listen on all addresses. -urlbase = "https://cat2.example.com" # The base of URLs, if different from ``http://``. -statedir = "_caterva2/sub" # The directory where the service will place state files. -loglevel = "warning" # All service messages having this severity or worse will be logged. - -# Only one of these is allowed. It will be used by a subscriber invoked with no ID, or it may be used by other programs to find how to connect to a subscriber (``subscriber.url``). [subscriber] -urlbase = "https://cat2.example.com" -# ... other settings as above ... +statedir = "_caterva2/sub" +#http = "_caterva2/sub/uvicorn.socket" +http = "localhost:8002" +urlbase = "http://localhost:8002" +quota = "10G" +maxusers = 5 +register = true # allow users to register -# Common configuration of client programs. -# Only one of these is allowed for the moment. +# The client section defines the credentials for the client to authenticate +# against the subscriber. [client] -username = "" # If present and not empty, a name to be used to authenticate the user to the subscriber and get an authorization token. -password = "" # If present and not empty, a password for the previous user. +username = "" +password = "" diff --git a/caterva2/client.py b/caterva2/client.py index e47bb3a9..ffd7c5e0 100644 --- a/caterva2/client.py +++ b/caterva2/client.py @@ -3,6 +3,7 @@ import io import pathlib import sys +import warnings from collections.abc import Sequence from pathlib import PurePosixPath @@ -783,8 +784,6 @@ def get_roots(self): dict Dictionary mapping available root names to their details: - ``name``: the root name - - ``http``: the HTTP endpoint - - ``subscribed``: whether it is subscribed or not. Examples -------- @@ -796,35 +795,11 @@ def get_roots(self): >>> client.subscribe('b2tests') 'Ok' >>> roots_dict['b2tests'] - {'name': 'b2tests', 'http': 'localhost:8014', 'subscribed': True} + {'name': 'b2tests'} """ urlbase, _ = _format_paths(self.urlbase) return api_utils.get(f"{self.urlbase}/api/roots", auth_cookie=self.cookie, timeout=self.timeout) - def _get_root(self, name): - """ - Retrieves a specified root name. - - Parameters - ---------- - name : str - Name of the root to retrieve. - - Returns - ------- - Root - An instance of :class:`Root`. - - """ - if "/" in name: - raise ValueError("Root names cannot contain slashes") - # It is a root, subscribe to it - ret = self.subscribe(name) - if ret != "Ok": - roots = self.get_roots() - raise ValueError(f"Could not subscribe to root {name} (only {roots.keys()} available)") - return Root(self, name) - def get(self, path): """ Returns an object for the given path. @@ -857,11 +832,11 @@ def get(self, path): path = pathlib.PurePosixPath(path).as_posix() # Check if the path is a root or a file/dataset if "/" not in path: - return self._get_root(path) + return Root(self, path) + # If not a root, assume it's a file/dataset - root_name = path.split("/")[0] - root = self._get_root(root_name) - file_path = path[len(root_name) + 1 :] + root_name, file_path = path.split("/", 1) + root = Root(self, root_name) return root[file_path] def subscribe(self, root): @@ -886,12 +861,10 @@ def subscribe(self, root): >>> client.subscribe(root_name) 'Ok' >>> client.get_roots()[root_name] - {'name': 'h5numbers_j2k', 'http': 'localhost:8011', 'subscribed': True} + {'name': 'h5numbers_j2k'} """ - urlbase, root = _format_paths(self.urlbase, root) - return api_utils.post( - f"{self.urlbase}/api/subscribe/{root}", auth_cookie=self.cookie, timeout=self.timeout - ) + warnings.warn("subscribe() is deprecated, it does nothing, just remove the call") + return "Ok" def get_list(self, path): """ diff --git a/caterva2/clients/cli.py b/caterva2/clients/cli.py index b800fc83..79c21cfe 100644 --- a/caterva2/clients/cli.py +++ b/caterva2/clients/cli.py @@ -56,21 +56,8 @@ def cmd_roots(client, args): print(json.dumps(data)) return - for name, root in data.items(): - if root["subscribed"] is True: - print(f"{name} (subscribed)") - else: - print(name) - - -@handle_errors -def cmd_subscribe(client, args): - data = client.subscribe(args.root) - if args.json: - print(json.dumps(data)) - return - - print(data) + for name in data: + print(name) @handle_errors @@ -195,20 +182,13 @@ def main(): subparsers = parser.add_subparsers(required=True) # roots - help = "List all the available roots in a broker." + help = "List all the available roots." subparser = subparsers.add_parser("roots", help=help) subparser.add_argument("--json", action="store_true") subparser.set_defaults(func=cmd_roots) - # subscribe - help = "Request access to the datasets in a root." - subparser = subparsers.add_parser("subscribe", help=help) - subparser.add_argument("--json", action="store_true") - subparser.add_argument("root") - subparser.set_defaults(func=cmd_subscribe) - # list - help = "List all the available datasets in a root. Needs to be subscribed to the root." + help = "List all the available datasets in a root." subparser = subparsers.add_parser("list", aliases=["ls"], help=help) subparser.add_argument("--json", action="store_true") subparser.add_argument("root") diff --git a/caterva2/models.py b/caterva2/models.py index 1e3f4ccc..51a85717 100644 --- a/caterva2/models.py +++ b/caterva2/models.py @@ -105,18 +105,7 @@ class File(pydantic.BaseModel): class Root(pydantic.BaseModel): name: str - http: str - subscribed: bool | None = None # Used only by the subscriber program - - -class Broker(pydantic.BaseModel): - roots: dict[str, Root] - - -class Publisher(pydantic.BaseModel): - etags: dict[str, str] class Subscriber(pydantic.BaseModel): - roots: dict[str, Root] - etags: dict[str, str] + pass diff --git a/caterva2/services/bro.py b/caterva2/services/bro.py deleted file mode 100644 index c0d70d98..00000000 --- a/caterva2/services/bro.py +++ /dev/null @@ -1,69 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - - -# Requirements -from fastapi import FastAPI -from fastapi.routing import APIRouter -from fastapi_websocket_pubsub import PubSubEndpoint - -# Project -from caterva2 import models, utils -from caterva2.services import srv_utils - -# State -database = None - - -# API -app = FastAPI() - - -@app.get("/api/roots", response_model_exclude_none=True) -async def get_roots() -> dict[str, models.Root]: - return database.roots - - -@app.post("/api/roots") -async def post_roots(root: models.Root) -> models.Root: - database.roots[root.name] = root - database.save() - await endpoint.publish(["@new"], root) - return root - - -# Pub/Sub interface -router = APIRouter() -endpoint = PubSubEndpoint() -endpoint.register_route(router) -app.include_router(router) - - -def main(): - conf = utils.get_conf("broker") - parser = utils.get_parser( - http=conf.get(".http", "localhost:8000"), - loglevel=conf.get(".loglevel", "warning"), - statedir=conf.get(".statedir", "_caterva2/bro"), - ) - args = utils.run_parser(parser) - - # Init database - # roots = {name: } - statedir = args.statedir.resolve() - global database - database = srv_utils.Database(statedir / "db.json", models.Broker(roots={})) - print(database.data) - - # Run - srv_utils.uvicorn_run(app, args) - - -if __name__ == "__main__": - main() diff --git a/caterva2/services/dirroot.py b/caterva2/services/dirroot.py deleted file mode 100644 index 763e037a..00000000 --- a/caterva2/services/dirroot.py +++ /dev/null @@ -1,204 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - -import io -import os -import pathlib -from collections.abc import AsyncIterator, Callable, Collection, Iterator - -try: - from typing import Self -except ImportError: # Python < 3.11 - from typing import TypeVar - Self = TypeVar('Self', bound='PubRoot') # noqa: F821 - -# Requirements -import blosc2 -import pydantic -import watchfiles - -# Project -from caterva2.services import pubroot, srv_utils - - -class DirectoryRoot: - """Represents a publisher root which keeps datasets as files - in a directory. - """ - - Path = pubroot.PubRoot.Path - - @classmethod - def get_maker(cls, target: str) -> Callable[[], Self] | None: - try: - path = pathlib.Path(target) - if not path.is_dir(): - return None - except Exception: - return None - return lambda: cls(path) - - def __init__(self, path: pathlib.Path): - abspath = path.resolve(strict=True) - # Force an error for non-dirs or non-readable dirs. - next(abspath.iterdir()) - - self.abspath = abspath - - def walk_dsets(self) -> Iterator[Path]: - return (self.Path(p.relative_to(self.abspath)) - for p in self.abspath.glob('**/*') - if not p.is_dir()) - - def _rel_to_abs(self, relpath: Path) -> pathlib.Path: - if relpath.is_absolute(): - raise ValueError(f"path is not relative: {str(relpath)!r}") - # ``.`` is removed on path instantiation, no need to check for it. - if os.path.pardir in relpath.parts: - raise ValueError(f"{str(os.path.pardir)!r} not allowed " - f"in path: {str(relpath)!r}") - abspath = self.abspath / relpath - if not abspath.is_file(): - raise pubroot.NoSuchDatasetError(relpath) - return abspath - - def exists_dset(self, relpath: Path) -> bool: - try: - abspath = self._rel_to_abs(relpath) - except pubroot.NoSuchDatasetError: - return False - return abspath.is_file() - - def get_dset_etag(self, relpath: Path) -> str: - abspath = self._rel_to_abs(relpath) - stat = abspath.stat() - return f'{stat.st_mtime}:{stat.st_size}' - - def get_dset_meta(self, relpath: Path) -> pydantic.BaseModel: - abspath = self._rel_to_abs(relpath) - return srv_utils.read_metadata(abspath) - - def get_dset_chunk(self, relpath: Path, nchunk: int) -> bytes: - abspath = self._rel_to_abs(relpath) - b2dset = blosc2.open(abspath) - schunk = getattr(b2dset, 'schunk', b2dset) - if nchunk > schunk.nchunks: - raise pubroot.NoSuchChunkError(nchunk) - return schunk.get_chunk(nchunk) - - def open_dset_raw(self, relpath: Path) -> io.RawIOBase: - abspath = self._rel_to_abs(relpath) - return open(abspath, 'rb') - - async def awatch_dsets(self) -> AsyncIterator[Collection[Path]]: - async for changes in watchfiles.awatch(self.abspath): - relpaths = { - self.Path(pathlib.Path(abspath).relative_to(self.abspath)) - for change, abspath in changes} - yield relpaths - - -pubroot.register_root_class(DirectoryRoot) - - -def create_example_root(path): - """Create an example Caterva2 directory to be used as a root.""" - import pathlib - - import blosc2 - import numpy as np - - # The examples come from ``SPECS.md`` and ``root-example`` content. - path = pathlib.Path(path) - path.mkdir(parents=True) - - with open(path / "README.md", "w") as f: - f.write("# Header example\n" - "This is a simple example,\n" - "\n" - "with several lines,\n" - "\n" - "for showing purposes.\n") - - # A SChunk containing a data buffer. - blosc2.SChunk(chunksize=100, data=b"Hello world!" * 100, - urlpath=path / "ds-hello.b2frame", mode="w") - - # A 1D array (int64). - a = np.arange(1000, dtype="int64") - blosc2.asarray(a, chunks=(100,), blocks=(10,), - urlpath=path / "ds-1d.b2nd", mode="w") - - # A 1D array (6-byte strings). - a = np.array([b"foobar"] * 1000) - blosc2.asarray(a, chunks=(100,), blocks=(10,), - urlpath=path / "ds-1d-b.b2nd", mode="w") - - # A 1D array (dtype with fields). - a = np.empty(1000, dtype=[("a", "int32"), ("b", "float64"), ("c", "S10"), ("d", "?")]) - a["a"] = np.arange(1000, dtype="int32") - a["b"] = np.linspace(0, 1, 1000, dtype="float64") - a["c"] = np.array([f"foobar{i}" for i in range(1000)], dtype="S10") - # A field with random booleans - a["d"] = np.random.default_rng().choice([True, False], 1000) - blosc2.asarray(a, chunks=(100,), blocks=(10,), - urlpath=path / "ds-1d-fields.b2nd", mode="w") - - # A 2D array with 2 fields - shape = (100, 200) - npa = np.linspace(0, 1, np.prod(shape), dtype=np.float32).reshape(shape) - npb = np.linspace(1, 2, np.prod(shape), dtype=np.float64).reshape(shape) - nps = np.empty(shape, dtype=[("a", npa.dtype), ("b", npb.dtype)]) - nps["a"] = npa - nps["b"] = npb - s = blosc2.asarray(nps, mode="w", urlpath=path / "ds-2d-fields.b2nd") - a = blosc2.NDField(s, "a") - b = blosc2.NDField(s, "b") - - # A scalar (string) with variable-length metalayers (user attributes). - a = np.str_("foobar") - b = blosc2.asarray(a, urlpath=path / "ds-sc-attr.b2nd", mode="w") - for k, v in {'a': 1, 'b': "foo", 'c': 123.456}.items(): - b.schunk.vlmeta[k] = v - - (path / "dir1").mkdir() - - # A 2D array (uint16). - a = np.arange(200, dtype="uint16").reshape(10, 20) - blosc2.asarray(a, chunks=(5, 5), blocks=(2, 3), - urlpath=path / "dir1/ds-2d.b2nd", mode="w") - - # A 3D array (float32). - a = np.arange(60, dtype="float32").reshape(3, 4, 5) - blosc2.asarray(a, chunks=(2, 3, 4), blocks=(2, 2, 2), - urlpath=path / "dir1/ds-3d.b2nd", mode="w") - - (path / "dir2").mkdir() - - # A 4D array (complex128). - a = np.arange(120, dtype="complex128").reshape(2, 3, 4, 5) - blosc2.asarray(a + a * 1j, chunks=(1, 2, 3, 4), blocks=(1, 2, 2, 2), - urlpath=path / "dir2/ds-4d.b2nd", mode="w") - - -def main(): - import os - import sys - try: - _, c2dpath = sys.argv - except ValueError: - prog = os.path.basename(sys.argv[0]) - print(f"Usage: {prog} CATERVA2_DIR", file=sys.stderr) - sys.exit(1) - create_example_root(c2dpath) - print(f"Created example Caterva2 root: {c2dpath!r}", file=sys.stderr) - - -if __name__ == '__main__': - main() diff --git a/caterva2/services/hdf5root.py b/caterva2/services/hdf5root.py deleted file mode 100644 index f8710293..00000000 --- a/caterva2/services/hdf5root.py +++ /dev/null @@ -1,260 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - -import functools -import io -import logging -import pathlib -import re -from collections.abc import AsyncIterator, Callable, Collection, Iterator, Mapping - -try: - from typing import Self -except ImportError: # Python < 3.11 - from typing import TypeVar - - Self = TypeVar("Self", bound="PubRoot") # noqa: F821 - -# Requirements -import h5py -import pydantic -import watchfiles - -# Project -from caterva2 import hdf5 -from caterva2.services import pubroot, srv_utils - -_MAX_CACHED_CHUNKERS = 32 -"""Maximum number of dataset chunkers to keep in per-instance LRU cache.""" - - -class HDF5Root: - Path = pubroot.PubRoot.Path - - @classmethod - def get_maker(cls, target: str) -> Callable[[], Self] | None: - try: - path = pathlib.Path(target) - if not h5py.is_hdf5(path): - return None - except Exception: - return None - return lambda: cls(path) - - def __init__(self, path: pathlib.Path): - self.h5file = h5py.File(path, mode="r") - - # There must be one cached function per instance, - # so that it can be reset individually. - # This means that just ``@functools.(lru_)cache`` - # on e.g. ``_b2args_from_h5dset(self, dset)`` is not enough - # (there would be a single cache shared by all instances). - - @functools.cached_property - def _b2args_from_h5dset(self): - @functools.cache # TODO: limit size? - def _getb2args(dset: h5py.Dataset) -> Mapping[str, object]: - return hdf5.b2args_from_h5dset(dset) - - return _getb2args - - @functools.cached_property - def _b2attrs_from_h5dset(self): - @functools.cache # TODO: limit size? - def _getb2attrs(dset: h5py.Dataset) -> Mapping[str, object]: - return hdf5.b2attrs_from_h5dset(dset) - - return _getb2attrs - - @functools.cached_property - def _b2chunkers_from_h5dset(self): - @functools.lru_cache(maxsize=_MAX_CACHED_CHUNKERS) # only hot datasets - def _getb2chunkers(dset: h5py.Dataset) -> (Callable[[int], bytes], Callable[[], Iterator[bytes]]): - b2_args = self._b2args_from_h5dset(dset) - return hdf5.b2chunkers_from_h5dset(dset, b2_args) - - return _getb2chunkers - - def _clear_caches(self): - self._b2args_from_h5dset.cache_clear() - self._b2attrs_from_h5dset.cache_clear() - self._b2chunkers_from_h5dset.cache_clear() - - def walk_dsets(self) -> Iterator[Path]: - # TODO: either iterate (without accumulation) or cache - dsets = [] - warn = logging.getLogger().warning - - def visitor(name, node): - if not _is_dataset(node): - if not isinstance(node, h5py.Group): - warn("skipping incompatible HDF5 dataset: %r", name) - return - # TODO: handle array / frame / (compressed) file distinctly - dsets.append(self.Path(f"{name}.b2nd")) - - self.h5file.visititems(visitor) - yield from iter(dsets) - - def _path_to_dset(self, relpath: Path) -> h5py.Dataset: - name = re.sub(r"\.b2(nd|frame)?$", "", str(relpath)) - node = self.h5file.get(name) - if node is None or not _is_dataset(node): - raise pubroot.NoSuchDatasetError(relpath) - return node - - def exists_dset(self, relpath: Path) -> bool: - try: - self._path_to_dset(relpath) - except pubroot.NoSuchDatasetError: - return False - return True - - def get_dset_etag(self, relpath: Path) -> str: - dset = self._path_to_dset(relpath) - # All datasets have the modification time of their file. - h5path = pathlib.Path(self.h5file.filename) - stat = h5path.stat() - return f"{stat.st_mtime}:{dset.nbytes}" - - def get_dset_meta(self, relpath: Path) -> pydantic.BaseModel: - dset = self._path_to_dset(relpath) - b2_args = self._b2args_from_h5dset(dset) - b2_attrs = self._b2attrs_from_h5dset(dset) - b2_array = hdf5.b2uninit_from_h5dset(dset, b2_args, b2_attrs) - return srv_utils.read_metadata(b2_array) - - def get_dset_chunk(self, relpath: Path, nchunk: int) -> bytes: - dset = self._path_to_dset(relpath) - b2getchunk, _ = self._b2chunkers_from_h5dset(dset) - try: - return b2getchunk(nchunk) - except IndexError as ie: - raise pubroot.NoSuchChunkError(*ie.args) from ie - - def open_dset_raw(self, relpath: Path) -> io.RawIOBase: - # TODO: handle array / frame / (compressed) file distinctly - raise NotImplementedError("cannot read raw contents of array dataset") - - async def awatch_dsets(self) -> AsyncIterator[Collection[Path]]: - h5path = self.h5file.filename - old_dsets = set(self.walk_dsets()) - async for _ in watchfiles.awatch(h5path): - self._clear_caches() - self.h5file.close() - self.h5file = h5py.File(h5path, mode="r") - # All datasets are supposed to change along with their file. - cur_dsets = set(self.walk_dsets()) - # Old datasets are included in case any of them disappeared. - yield old_dsets | cur_dsets - old_dsets = cur_dsets - - -pubroot.register_root_class(HDF5Root) - - -def _is_dataset(node: h5py.Group | h5py.Dataset) -> bool: - return isinstance(node, h5py.Dataset) and hdf5.h5dset_is_compatible(node) - - -def create_example_root(path): - """Create an example HDF5 file to be used as a root.""" - import h5py - import numpy - from hdf5plugin import Blosc2 as B2Comp - - with h5py.File(path, "x") as h5f: - h5f.create_dataset("/scalar", data=123.456) - h5f.create_dataset("/string", data=numpy.bytes_("Hello world!")) - - a = numpy.arange(100, dtype="uint8") - h5f.create_dataset("/arrays/1d-raw", data=a) - - h5f["/arrays/soft-link"] = h5py.SoftLink("/arrays/1d-raw") - - a = numpy.array([b"foobar"] * 100) - h5f.create_dataset("/arrays/1ds-blosc2", data=a, chunks=(50,), **B2Comp()) - - a = numpy.arange(100, dtype="complex128").reshape(10, 10) - a = a + a * 1j - h5f.create_dataset("/arrays/2d-nochunks", data=a, chunks=None) - - a = numpy.arange(100, dtype="complex128").reshape(10, 10) - a = a + a * 1j - h5f.create_dataset("/arrays/2d-gzip", data=a, chunks=(4, 4), compression="gzip") - - a = numpy.arange(1000, dtype="uint8").reshape(10, 10, 10) - h5f.create_dataset( - "/arrays/3d-blosc2", - data=a, - chunks=(4, 10, 10), - **B2Comp(cname="lz4", clevel=7, filters=B2Comp.BITSHUFFLE), - ) - - a = numpy.linspace(-1, 2, 1000).reshape(10, 10, 10) - h5f.create_dataset( - "/arrays/3d-blosc2-a", - data=a, - chunks=(4, 10, 10), - **B2Comp(cname="lz4", clevel=7, filters=B2Comp.BITSHUFFLE), - ) - - a = numpy.linspace(-1, 2, 1000).reshape(10, 10, 10) - h5f.create_dataset( - "/arrays/3d-blosc2-b", - data=a, - chunks=(2, 5, 10), - **B2Comp(cname="blosclz", clevel=7, filters=B2Comp.SHUFFLE), - ) - - h5f.create_dataset("/arrays/array-dtype", dtype=numpy.dtype(("float64", (4,))), shape=(10,)) - - ds = h5f.create_dataset("/attrs", data=0) - a = numpy.arange(4, dtype="uint8").reshape(2, 2) - for k, v in { - "Int": 42, - "IntT": numpy.int16(42), - "Bin": b"foo", - "BinT": numpy.bytes_(b"foo"), - "Str": "bar", # StrT=numpy.str_("bar"), - "Arr": a.tolist(), - "ArrT": a, - "NilBin": h5py.Empty("|S4"), - # NilStr=h5py.Empty('|U4'), - "NilInt": h5py.Empty("uint8"), - }.items(): - ds.attrs[k] = v - - h5f.create_dataset("/arrays/empty", data=h5py.Empty("float64")) - - h5f.create_dataset("/arrays/compound-dtype", dtype=numpy.dtype("uint8,float64"), shape=(10,)) - - a = numpy.arange(1, dtype="uint8").reshape((1,) * 23) - h5f.create_dataset("/unsupported/too-many-dimensions", data=a) - - # TODO: This could be supported by mapping the vlstring dataset to an NDArray with shape=() and dtype="u1" - h5f.create_dataset("/unsupported/vlstring", data="Hello world!") - - -def main(): - import os - import sys - - try: - _, h5fpath = sys.argv - except ValueError: - prog = os.path.basename(sys.argv[0]) - print(f"Usage: {prog} HDF5_FILE", file=sys.stderr) - sys.exit(1) - create_example_root(h5fpath) - print(f"Created example HDF5 root: {h5fpath!r}", file=sys.stderr) - - -if __name__ == "__main__": - main() diff --git a/caterva2/services/pub.py b/caterva2/services/pub.py deleted file mode 100644 index 757bb342..00000000 --- a/caterva2/services/pub.py +++ /dev/null @@ -1,230 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - -import asyncio -import contextlib -import logging - -# Requirements -import blosc2 -from fastapi import FastAPI, Response, responses - -import caterva2.services.dirroot - -# Project -from caterva2 import api_utils, models, utils -from caterva2.services import pubroot, srv_utils - -with contextlib.suppress(ImportError): - import caterva2.services.hdf5root # noqa: F401 - - -logger = logging.getLogger("pub") - -# Configuration -broker = None -name = None -proot = None -nworkers = 1 - -# State -cache = None -client = None -database = None # instance - - -async def worker(queue): - while True: - relpath = await queue.get() - with utils.log_exception(logger, "Publication failed"): - assert isinstance(relpath, proot.Path) - key = str(relpath) - if proot.exists_dset(relpath): - print("UPDATE", relpath) - # Load metadata - if relpath.suffix in {".b2frame", ".b2nd"}: - metadata = proot.get_dset_meta(relpath) - else: - # Compress regular files in publisher's cache - with proot.open_dset_raw(relpath) as f: - data = f.read() - b2path = cache / f"{relpath}.b2" - srv_utils.compress(data, b2path) - metadata = srv_utils.read_metadata(b2path) - - # Publish - metadata = metadata.model_dump() - data = {"path": str(relpath), "metadata": metadata} - await client.publish(name, data=data) - # Update database - database.etags[key] = proot.get_dset_etag(relpath) - database.save() - else: - print("DELETE", relpath) - data = {"path": str(relpath)} - await client.publish(name, data=data) - # Update database - if key in database.etags: - del database.etags[key] - database.save() - - queue.task_done() - - -async def watch_root(queue): - # On start, notify the network about changes to the datasets, changes done since the - # last run. - etags = database.etags.copy() - for relpath in proot.walk_dsets(): - key = str(relpath) - val = etags.pop(key, None) - if val != proot.get_dset_etag(relpath): - queue.put_nowait(relpath) - - # The etags left are those that were deleted - for key in etags: - relpath = proot.Path(key) - queue.put_nowait(relpath) - del database.etags[key] - database.save() - - # Watch root for changes - async for changes in proot.awatch_dsets(): - for relpath in changes: - queue.put_nowait(relpath) - - print("THIS SHOULD BE PRINTED ON CTRL+C") - - -@contextlib.asynccontextmanager -async def lifespan(app: FastAPI): - # Connect to broker - global client - client = srv_utils.start_client(f"ws://{broker}/pubsub") - - # Create queue and start workers - queue = asyncio.Queue() - tasks = [] - for _ in range(nworkers): - task = asyncio.create_task(worker(queue)) - tasks.append(task) - - # Watch dataset files (must wait before publishing) - await client.wait_until_ready() - watch_task = asyncio.create_task(watch_root(queue)) - - yield - - # Cancel watch task - watch_task.cancel() - - # Cancel worker tasks - for task in tasks: - task.cancel() - await asyncio.gather(*tasks, return_exceptions=True) - - # Disconnect from broker - await srv_utils.disconnect_client(client) - - -app = FastAPI(lifespan=lifespan) - - -@app.get("/api/list") -async def get_list(): - return list(proot.walk_dsets()) - - -@app.get("/api/info/{path:path}") -async def get_info( - path: str, - response: Response, - if_none_match: srv_utils.HeaderType = None, -): - relpath = proot.Path(path) - srv_utils.check_dset_path(proot, relpath) - - # Check etag - etag = database.etags[str(relpath)] - if if_none_match == etag: - return Response(status_code=304) - - if relpath.suffix in {".b2frame", ".b2nd"}: - meta = proot.get_dset_meta(relpath) - else: - b2path = srv_utils.get_abspath(cache, f"{relpath}.b2") - meta = srv_utils.read_metadata(b2path) - - # Return - response.headers["Etag"] = etag - return meta - - -@app.get("/api/download/{path:path}") -async def get_download(path: str, nchunk: int = -1): - if nchunk < 0: - srv_utils.raise_bad_request("Chunk number required") - - relpath = proot.Path(path) - srv_utils.check_dset_path(proot, relpath) - - if relpath.suffix in {".b2frame", ".b2nd"}: - chunk = proot.get_dset_chunk(relpath, nchunk) - else: - b2path = cache / (f"{relpath}.b2") - schunk = blosc2.open(b2path) - chunk = schunk.get_chunk(nchunk) - - downloader = srv_utils.iterchunk(chunk) - return responses.StreamingResponse(downloader) - - -def main(): - conf = utils.get_conf("publisher", allow_id=True) - _stdir = "_caterva2/pub" + (f".{conf.id}" if conf.id else "") - parser = utils.get_parser( - broker=conf.get("broker.http", "localhost:8000"), - http=conf.get(".http", "localhost:8001"), - loglevel=conf.get(".loglevel", "warning"), - statedir=conf.get(".statedir", _stdir), - id=conf.id, - ) - parser.add_argument("name", nargs="?", default=conf.get(".name")) - parser.add_argument("root", nargs="?", default=conf.get(".root", "data")) - args = utils.run_parser(parser) - if args.name is None: # because optional positional arg w/o conf default - raise RuntimeError("root name was not specified in configuration nor in arguments") - - # Global configuration - global broker, name, proot - broker = args.broker - name = args.name - proot = pubroot.make_root(args.root) - - # Init cache - global cache - statedir = args.statedir.resolve() - cache = statedir / "cache" - cache.mkdir(exist_ok=True, parents=True) - - # Init database - global database - model = models.Publisher(etags={}) - database = srv_utils.Database(statedir / "db.json", model) - - # Register - data = {"name": name, "http": args.http} - api_utils.post("/api/roots", json=data, server=args.broker) - - # Run - srv_utils.uvicorn_run(app, args) - - -if __name__ == "__main__": - main() diff --git a/caterva2/services/pubroot.py b/caterva2/services/pubroot.py deleted file mode 100644 index 8b237991..00000000 --- a/caterva2/services/pubroot.py +++ /dev/null @@ -1,134 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - -"""Publisher root classes. - -This includes an abstract `PubRoot` class defining the interface that concrete -classes must implement to support different publisher root sources. New -classes may be registered with the `register_root_class()` function. - -The `make_root()` function, given a target string argument, tries to find the -adequate class that understands the target and can create a publisher root -instance from it. -""" - -import io -import pathlib -from abc import ABC, abstractmethod -from collections.abc import AsyncIterator, Callable, Collection, Iterator - -try: - from typing import Self -except ImportError: # Python < 3.11 - from typing import TypeVar - Self = TypeVar('Self', bound='PubRoot') - -# Requirements -import pydantic - - -class NoSuchDatasetError(LookupError): - """The given dataset does not exist.""" - - -class NoSuchChunkError(IndexError): - """The given chunk does not exist in the dataset.""" - - -class PubRoot(ABC): - """Abstract class that represents a publisher root.""" - - """The class of dataset (relative) paths.""" - Path = pathlib.PurePosixPath - - @classmethod - @abstractmethod - def get_maker(cls, target: str) -> Callable[[], Self] | None: - """Return a callable that returns a root for the given `target`. - - If `target` cannot be used to create an instance of this class, - return `None`, but no exception should be raised. - """ - - @abstractmethod - def walk_dsets(self) -> Iterator[Path]: - """Iterate over the relative paths of datasets in this root.""" - - @abstractmethod - def exists_dset(self, relpath: Path) -> bool: - """Does the named dataset exist?""" - - @abstractmethod - def get_dset_etag(self, relpath: Path) -> str: - """Get a string that varies if the named dataset is modified. - - Raise `NoSuchDatasetError` if the dataset does not exist. - """ - - @abstractmethod - def get_dset_meta(self, relpath: Path) -> pydantic.BaseModel: - """Get the metadata of the named dataset. - - Raise `NoSuchDatasetError` if the dataset does not exist. - """ - - @abstractmethod - def get_dset_chunk(self, relpath: Path, nchunk: int) -> bytes: - """Get compressed chunk with index `nchunk` of the named dataset. - - Raise `NoSuchDatasetError` if the dataset does not exist. - Raise `NoSuchChunkError` if the chunk does not exist. - """ - - @abstractmethod - def open_dset_raw(self, relpath: Path) -> io.BufferedIOBase: - """Get a byte reader for the raw contents of the named dataset. - - Raise `NoSuchDatasetError` if the dataset does not exist. - """ - - @abstractmethod - async def awatch_dsets(self) -> AsyncIterator[Collection[Path]]: - """Yield a set of datasets that have been modified.""" - - -_registered_classes = [] - - -def register_root_class(cls: type) -> bool: - """Add a publisher root class to the registry. - - This also registers the class as a virtual subclass of `PubRoot`. - - Return whether the class was added or not (because it was already). - """ - if cls in _registered_classes: - return False - PubRoot.register(cls) - _registered_classes.append(cls) - return True - - -class UnsupportedRootError(Exception): - """No publisher root class supports the given target.""" - - -def make_root(target: str) -> PubRoot: - """Return a publisher root instance for the given `target`. - - If no registered publisher root class supports the given `target`, - raise `UnsupportedRootError`. - """ - for cls in _registered_classes: - maker = cls.get_maker(target) - if maker is not None: - return maker() - else: - raise UnsupportedRootError(f"no publisher root class could be used " - f"for target: {target!r}") diff --git a/caterva2/services/srv_utils.py b/caterva2/services/srv_utils.py index a7779cba..b8be6d16 100644 --- a/caterva2/services/srv_utils.py +++ b/caterva2/services/srv_utils.py @@ -20,7 +20,6 @@ # Requirements import blosc2 import fastapi -import fastapi_websocket_pubsub import safer import uvicorn from fastapi_users.exceptions import UserNotExists @@ -228,23 +227,6 @@ def operands_as_paths(operands, cache, personal, shared, public): } -# -# Pub/Sub helpers -# - - -def start_client(url): - client = fastapi_websocket_pubsub.PubSubClient() - client.start_client(url) - return client - - -async def disconnect_client(client, timeout=5): - if client is not None: - # If the broker is down client.disconnect hangs, wo we wrap it in a timeout - await asyncio.wait_for(client.disconnect(), timeout) - - # # HTTP server helpers # @@ -278,16 +260,6 @@ def get_abspath(root, path, may_not_exist=False): return abspath -def check_dset_path(proot, path): - try: - exists = proot.exists_dset(path) - except ValueError: - raise_bad_request(f"Invalid path {path}") - else: - if not exists: - raise_not_found() - - def uvicorn_run(app, args, root_path=""): http = args.http if http.uds: diff --git a/caterva2/services/sub.py b/caterva2/services/sub.py index 265390c0..d0198d2b 100644 --- a/caterva2/services/sub.py +++ b/caterva2/services/sub.py @@ -33,11 +33,9 @@ import dotenv import fastapi import furl -import httpx import markdown import nbconvert import nbformat -import numpy as np import PIL.Image # FastAPI @@ -60,7 +58,6 @@ logger = logging.getLogger("sub") # State -clients = {} # topic: locks = {} mimetypes.add_type("text/markdown", ".md") # Because in macOS this is not by default @@ -72,140 +69,6 @@ def guess_type(path): return mimetype -class PubSourceDataset: - """ - Class for getting chunks from a dataset on a publisher service. - """ - - def __init__(self, abspath, path, metadata=None): - self.path = pathlib.Path(path) - if metadata is not None: - suffix = abspath.suffix - if suffix == ".b2nd": - metadata = models.Metadata(**metadata) - self._shape = metadata.shape - self._chunks = metadata.chunks - self._blocks = metadata.blocks - dtype = metadata.dtype - # Sometimes dtype is a tuple (e.g. (' bytes: - return self._get_chunk(nchunk) - - async def aget_chunk(self, nchunk: int) -> bytes: - return await self._aget_chunk(nchunk) - - -# Class representing a SChunk in a publisher -class PubSCDataset(blosc2.ProxySource, PubSourceDataset): - @property - def typesize(self): - return self._typesize - - @property - def chunksize(self): - return self._chunksize - - @property - def nbytes(self): - return self._nbytes - - @property - def cparams(self): - return self._cparams - - def get_chunk(self, nchunk: int) -> bytes: - return self._get_chunk(nchunk) - - async def aget_chunk(self, nchunk: int) -> bytes: - return await self._aget_chunk(nchunk) - - -# Factory function for creating a proxy for a dataset in publisher -def PubDataset(abspath, path, metadata=None): - dataset = PubSourceDataset(abspath, path, metadata) - # By using __new__() and updating the internal dict of the instance, - # we can return the right class avoiding calling PubSourceDataset.__init__ again - if hasattr(dataset, "_shape"): - # return PubNDDataset(abspath, path, metadata) - instance = PubNDDataset.__new__(PubNDDataset) - else: - # return PubSCDataset(abspath, path, metadata) - instance = PubSCDataset.__new__(PubSCDataset) - instance.__dict__.update(dataset.__dict__) - return instance - - def get_disk_usage(): exclude = {"db.json", "db.sqlite"} return sum(path.stat().st_size for path, _ in utils.walk_files(settings.statedir, exclude=exclude)) @@ -244,42 +107,6 @@ def make_url(request, name, query=None, **path_params): return settings.urlbase + url -async def new_root(data, topic): - logger.info(f"NEW root {topic} {data=}") - root = models.Root(**data) - settings.database.roots[root.name] = root - settings.database.save() - - -async def updated_dataset(data, topic): - name = topic - relpath = data["path"] - - rootdir = settings.cache / name - abspath = rootdir / relpath - metadata = data.get("metadata") - if metadata is None: - if abspath.suffix not in {".b2nd", ".b2frame"}: - abspath = pathlib.Path(f"{abspath}.b2") - if abspath.is_file(): - abspath.unlink() - else: - key = f"{name}/{relpath}" - init_b2(abspath, key, metadata) - - -def init_b2(abspath, path, metadata): - dataset = PubDataset(abspath, path, metadata) - # TODO: not sure if this would prevent some kind of update in dataset. @jdavid? - # if os.path.exists(dataset.abspath): - # return - schunk_meta = metadata.get("schunk", metadata) - vlmeta = {} - for k, v in schunk_meta["vlmeta"].items(): - vlmeta[k] = v - blosc2.Proxy(dataset, urlpath=dataset.abspath, vlmeta=vlmeta, caterva2_env=True) - - def open_b2(abspath, path): """ Open a Blosc2 dataset. @@ -287,125 +114,59 @@ def open_b2(abspath, path): Return a Proxy if the dataset is in a publisher, or the LazyExpr or Blosc2 container otherwise. """ - if pathlib.Path(path).parts[0] in {"@personal", "@shared", "@public"}: - container = blosc2.open(abspath) - vlmeta = container.schunk.vlmeta if hasattr(container, "schunk") else container.vlmeta - if isinstance(container, blosc2.LazyExpr): - # Open the operands properly - operands = container.operands - for key, value in operands.items(): - if value is None: - raise ValueError(f'Missing operand "{key}"') - metaval = value.schunk.meta if hasattr(value, "schunk") else {} - vlmetaval = value.schunk.vlmeta - if "proxy-source" in metaval or ("_ftype" in vlmetaval and vlmetaval["_ftype"] == "hdf5"): - # Save operand as Proxy, see blosc2.open doc for more info. - # Or, it can be an HDF5 dataset too (which should be handled in the next call) - relpath = srv_utils.get_relpath(value) - operands[key] = open_b2(value.schunk.urlpath, relpath) - - if not hasattr(container, "_where_args"): - # If the container does not have _where_args, it is a LazyExpr - # and we can return it directly. - return container - - # Repeat the operation for where args (for properly handling proxies) - where_args = container._where_args - for key, value in where_args.items(): - if value is None: - raise ValueError(f'Missing operand "{key}"') - metaval = value.schunk.meta if hasattr(value, "schunk") else {} - vlmetaval = value.schunk.vlmeta if hasattr(value, "schunk") else {} - if "proxy-source" in metaval or ("_ftype" in vlmetaval and vlmetaval["_ftype"] == "hdf5"): - relpath = srv_utils.get_relpath(value) - value = open_b2(value.schunk.urlpath, relpath) - where_args[key] = value - elif isinstance(value, blosc2.LazyExpr): - # Properly open the operands (to e.g. find proxies) - for opkey, opvalue in value.operands.items(): - if isinstance(opvalue, blosc2.LazyExpr): - continue - relpath = srv_utils.get_relpath(opvalue) - value.operands[opkey] = open_b2(opvalue.schunk.urlpath, relpath) - - return container - - # Check if this is a file of a special type - elif "_ftype" in vlmeta and vlmeta["_ftype"] == "hdf5": - container = hdf5.HDF5Proxy(container) - # Set the number of threads for compression and decompression - container.cparams.nthreads = ncores - container.dparams.nthreads = ncores - return container + root = pathlib.Path(path).parts[0] + if root not in {"@personal", "@shared", "@public"}: + raise ValueError(f"Unexpected root={root}") - # Return Proxy - dataset = PubDataset(abspath, path) container = blosc2.open(abspath) - # No need to pass caterva2_env=True since _cache has already been created - return blosc2.Proxy(dataset, _cache=container) - - -# -# Internal API -# - - -def follow(name: str): - root = settings.database.roots.get(name) - if root is None: - return {name: "This dataset does not exist in the network"} - - if not root.subscribed: - root.subscribed = True - settings.database.save() - - # Create root directory in the cache - rootdir = settings.cache / name - if not rootdir.exists(): - rootdir.mkdir(exist_ok=True) - - # Get list of datasets - try: - data = api_utils.get("/api/list", server=root.http) - except httpx.ConnectError: - return None - - # Initialize the datasets in the cache - for relpath in data: - # If-None-Match header - key = f"{name}/{relpath}" - val = settings.database.etags.get(key) - headers = None if val is None else {"If-None-Match": val} - - # Call API - response = api_utils.get( - f"/api/info/{relpath}", - headers=headers, - server=root.http, - raise_for_status=False, - return_response=True, - ) - if response.status_code == 304: - continue - - response.raise_for_status() - metadata = response.json() - - # Save metadata and create Proxy - abspath = rootdir / relpath - init_b2(abspath, key, metadata) + vlmeta = container.schunk.vlmeta if hasattr(container, "schunk") else container.vlmeta + if isinstance(container, blosc2.LazyExpr): + # Open the operands properly + operands = container.operands + for key, value in operands.items(): + if value is None: + raise ValueError(f'Missing operand "{key}"') + metaval = value.schunk.meta if hasattr(value, "schunk") else {} + vlmetaval = value.schunk.vlmeta + if "proxy-source" in metaval or ("_ftype" in vlmetaval and vlmetaval["_ftype"] == "hdf5"): + # Save operand as Proxy, see blosc2.open doc for more info. + # Or, it can be an HDF5 dataset too (which should be handled in the next call) + relpath = srv_utils.get_relpath(value) + operands[key] = open_b2(value.schunk.urlpath, relpath) + + if not hasattr(container, "_where_args"): + # If the container does not have _where_args, it is a LazyExpr + # and we can return it directly. + return container - # Save etag - settings.database.etags[key] = response.headers["etag"] - settings.database.save() + # Repeat the operation for where args (for properly handling proxies) + where_args = container._where_args + for key, value in where_args.items(): + if value is None: + raise ValueError(f'Missing operand "{key}"') + metaval = value.schunk.meta if hasattr(value, "schunk") else {} + vlmetaval = value.schunk.vlmeta if hasattr(value, "schunk") else {} + if "proxy-source" in metaval or ("_ftype" in vlmetaval and vlmetaval["_ftype"] == "hdf5"): + relpath = srv_utils.get_relpath(value) + value = open_b2(value.schunk.urlpath, relpath) + where_args[key] = value + elif isinstance(value, blosc2.LazyExpr): + # Properly open the operands (to e.g. find proxies) + for opkey, opvalue in value.operands.items(): + if isinstance(opvalue, blosc2.LazyExpr): + continue + relpath = srv_utils.get_relpath(opvalue) + value.operands[opkey] = open_b2(opvalue.schunk.urlpath, relpath) - # Subscribe to changes in the dataset - if name not in clients: - client = srv_utils.start_client(f"ws://{settings.broker}/pubsub") - client.subscribe(name, updated_dataset) - clients[name] = client + return container - return None + # Check if this is a file of a special type + elif "_ftype" in vlmeta and vlmeta["_ftype"] == "hdf5": + container = hdf5.HDF5Proxy(container) + # Set the number of threads for compression and decompression + container.cparams.nthreads = ncores + container.dparams.nthreads = ncores + return container # @@ -461,50 +222,8 @@ async def lifespan(app: FastAPI): if user_login_enabled(): await db.create_db_and_tables(settings.statedir) - # Initialize roots from the broker - client = None - if settings.broker: - try: - data = api_utils.get("/api/roots", server=settings.broker) - except httpx.ConnectError: - logger.warning(f'Broker "{settings.broker}" not available') - else: - changed = False - # Deleted - d = list(settings.database.roots.items()) - for name, _root in d: - if name not in data: - del settings.database.roots[name] - changed = True - - # New or updated - for name, values in data.items(): - root = models.Root(**values) - if name not in settings.database.roots: - settings.database.roots[root.name] = root - changed = True - elif settings.database.roots[root.name].http != root.http: - settings.database.roots[root.name].http = root.http - changed = True - - if changed: - settings.database.save() - - # Follow the @new channel to know when a new root is added - client = srv_utils.start_client(f"ws://{settings.broker}/pubsub") - client.subscribe("@new", new_root) - - # Resume following - for path in settings.cache.iterdir(): - if path.is_dir(): - follow(path.name) - yield - # Disconnect from worker - if client is not None: - await srv_utils.disconnect_client(client) - # Visualize the size of a file on a compact and human-readable format def custom_filesizeformat(value): @@ -575,56 +294,18 @@ async def get_roots(user: db.User = Depends(optional_user)) -> dict: dict The dict of roots. """ - # Here we just return the roots that are known by the broker - # plus the special roots @personal, @shared and @public - roots = settings.database.roots.copy() - root = models.Root(name="@public", http="", subscribed=True) + # Here we just return the special roots @personal, @shared and @public + roots = {} + root = models.Root(name="@public") roots[root.name] = root if user: for name in ["@personal", "@shared"]: - root = models.Root(name=name, http="", subscribed=True) + root = models.Root(name=name) roots[root.name] = root return roots -def get_root(name): - root = settings.database.roots.get(name) - if root is None: - srv_utils.raise_not_found(f"{name} not known by the broker") - - return root - - -@app.post("/api/subscribe/{name}") -async def post_subscribe( - name: str, - user: db.User = Depends(optional_user), -): - """ - Subscribe to a root. - - Parameters - ---------- - name : str - The name of the root. - - Returns - ------- - str - 'Ok' if successful. - """ - if name == "@public": - pass - elif name in {"@personal", "@shared"}: - if not user: - raise srv_utils.raise_unauthorized(f"Subscribing to {name} requires authentication") - else: - get_root(name) - follow(name) - return "Ok" - - @app.get("/api/list/{path:path}") async def get_list( path: pathlib.Path, @@ -656,8 +337,7 @@ async def get_list( srv_utils.raise_not_found("@shared needs authentication") rootdir = settings.shared else: - root = get_root(root) - rootdir = settings.cache / root.name + raise ValueError(f"Unexpected root={root}") # List the datasets in root or directory directory = rootdir / pathlib.Path(*path.parts[1:]) @@ -734,39 +414,27 @@ def abspath_and_dataprep( given, or the whole data otherwise. """ parts = path.parts - if parts[0] == "@personal": - if not user: - raise fastapi.HTTPException(status_code=404) # NotFound + root = parts[0] + if root not in {"@personal", "@shared", "@public"}: + raise ValueError(f"Unexpected root={root}") + + if root in {"@personal", "@shared"} and not user: + raise fastapi.HTTPException(status_code=401) # Unauthorized + if root == "@personal": filepath = settings.personal / str(user.id) / pathlib.Path(*parts[1:]) abspath = srv_utils.cache_lookup(settings.personal, filepath, may_not_exist) - async def dataprep(): - pass - - elif parts[0] == "@shared": - if not user: - raise fastapi.HTTPException(status_code=404) # NotFound - + elif root == "@shared": filepath = settings.shared / pathlib.Path(*parts[1:]) abspath = srv_utils.cache_lookup(settings.shared, filepath, may_not_exist) - async def dataprep(): - pass - - elif parts[0] == "@public": + elif root == "@public": filepath = settings.public / pathlib.Path(*parts[1:]) abspath = srv_utils.cache_lookup(settings.public, filepath, may_not_exist) - async def dataprep(): - pass - - else: - filepath = settings.cache / path - abspath = srv_utils.cache_lookup(settings.cache, filepath, may_not_exist) - - async def dataprep(): - return await partial_download(abspath, path, slice_) + async def dataprep(): + pass return (abspath, dataprep) @@ -929,21 +597,20 @@ async def get_chunk( lock = locks.setdefault(path, asyncio.Lock()) async with lock: root = path.parts[0] - if root in {"@personal", "@shared", "@public"}: - if path in {"@personal", "@shared"} and not user: - raise fastapi.HTTPException(status_code=401) # Unauthorized - - container = open_b2(abspath, path) - if isinstance(container, blosc2.LazyArray): - # We do not support LazyUDF in Caterva2 yet. - # In case we do, this would have to be changed. - chunk = container.get_chunk(nchunk) - else: - schunk = getattr(container, "schunk", container) - chunk = schunk.get_chunk(nchunk) + if root not in {"@personal", "@shared", "@public"}: + raise ValueError(f"Unexpected root={root}") + + if root in {"@personal", "@shared"} and not user: + raise fastapi.HTTPException(status_code=401) # Unauthorized + + container = open_b2(abspath, path) + if isinstance(container, blosc2.LazyArray): + # We do not support LazyUDF in Caterva2 yet. + # In case we do, this would have to be changed. + chunk = container.get_chunk(nchunk) else: - sub_dset = PubDataset(abspath, path) - chunk = await sub_dset.aget_chunk(nchunk) + schunk = getattr(container, "schunk", container) + chunk = schunk.get_chunk(nchunk) downloader = srv_utils.iterchunk(chunk) return responses.StreamingResponse(downloader) @@ -1118,13 +785,18 @@ async def copy( raise srv_utils.raise_unauthorized("Copying files requires authentication") src, dst = payload.src, payload.dst + # src should start with a special root or known root - if not src.startswith(("@personal", "@shared", "@public")) and src not in settings.database.roots: - raise fastapi.HTTPException(status_code=400, detail="Only copying from existing roots is allowed") + if not src.startswith(("@personal", "@shared", "@public")): + raise fastapi.HTTPException( + status_code=400, + detail="Only copying from @personal or @shared or @public roots is allowed", + ) # dst should start with a special root if not dst.startswith(("@personal", "@shared", "@public")): raise fastapi.HTTPException( - status_code=400, detail="Only copying to @personal or @shared or @public roots is allowed" + status_code=400, + detail="Only copying to @personal or @shared or @public roots is allowed", ) namepath, destpath = pathlib.Path(src), pathlib.Path(dst) @@ -1160,9 +832,10 @@ def concatstackhelper(payload: models.ConcatStackPayload, user: db.User = Depend srcs, dst = payload.srcs, payload.dst # src should start with a special root or known root for src in srcs: - if not src.startswith(("@personal", "@shared", "@public")) and src not in settings.database.roots: + if not src.startswith(("@personal", "@shared", "@public")): raise fastapi.HTTPException( - status_code=400, detail="Only stacking/concatenating from existing roots is allowed" + status_code=400, + detail="Only stacking/concatenating from @personal or @shared or @public roots is allowed", ) # dst should start with a special root and if not try and massage it if not dst.startswith(("@personal", "@shared", "@public")): @@ -1758,7 +1431,6 @@ async def htmx_root_list( ): context = { "checked": roots, - "roots": sorted(settings.database.roots.values(), key=lambda x: x.name), "user": user, } return templates.TemplateResponse(request, "root_list.html", context) @@ -1774,9 +1446,7 @@ def _get_rootdir(user, root): elif root == "@public": return settings.public else: - if not get_root(root).subscribed: - follow(root) - return settings.cache / root + raise ValueError(f"Unexpected root={root}") return None @@ -2772,7 +2442,6 @@ def directory(abspath, relpath, content=None): if rootdir is not None: content.append(directory(rootdir, "@public")) - # TODO pub/sub roots: settings.database.roots.values() dir_abspath = rootdir.parent dir_relpath = "" else: @@ -2882,14 +2551,12 @@ def main(): # Parse command line arguments _stdir = "_caterva2/sub" + (f".{conf.id}" if conf.id else "") parser = utils.get_parser( - broker=conf.get("broker.http", ""), http=conf.get(".http", "localhost:8002"), loglevel=conf.get(".loglevel", "warning"), statedir=conf.get(".statedir", _stdir), id=conf.id, ) args = utils.run_parser(parser) - settings.broker = args.broker # Init cache settings.statedir = args.statedir.resolve() @@ -2911,7 +2578,7 @@ def main(): # app.mount("/personal", StaticFiles(directory=settings.personal), name="personal") # Init database - model = models.Subscriber(roots={}, etags={}) + model = models.Subscriber() settings.database = srv_utils.Database(settings.statedir / "db.json", model) # Register display plugins (delay module load) diff --git a/caterva2/services/templates/root_list.html b/caterva2/services/templates/root_list.html index eaa99f52..ea4478c2 100644 --- a/caterva2/services/templates/root_list.html +++ b/caterva2/services/templates/root_list.html @@ -11,12 +11,6 @@ Roots: - {% for root in roots %} - {% with name=root.name %} - {% include 'includes/root.html' %} - {% endwith %} - {% endfor %} - {% if user %} {% with name="@personal", upload=True %} {% set title="This special root contains the results of calculations done by the user" %} diff --git a/caterva2/tests/caterva2-login.toml b/caterva2/tests/caterva2-login.toml index 165957c7..885e5ae0 100644 --- a/caterva2/tests/caterva2-login.toml +++ b/caterva2/tests/caterva2-login.toml @@ -1,6 +1,3 @@ -[broker] -http = "localhost:8000" - [subscriber] login = true register = true diff --git a/caterva2/tests/caterva2-nologin.toml b/caterva2/tests/caterva2-nologin.toml index 513216e3..c466cd70 100644 --- a/caterva2/tests/caterva2-nologin.toml +++ b/caterva2/tests/caterva2-nologin.toml @@ -1,6 +1,3 @@ -[broker] -http = "localhost:8000" - [subscriber] login = false register = false diff --git a/caterva2/tests/conftest.py b/caterva2/tests/conftest.py index 9d355b87..47db77e1 100644 --- a/caterva2/tests/conftest.py +++ b/caterva2/tests/conftest.py @@ -8,7 +8,7 @@ import caterva2 as cat2 from .conf import configuration # noqa: F401 -from .files import examples_dir, examples_hdf5 # noqa: F401 +from .files import examples_dir # noqa: F401 from .services import services # noqa: F401 from .sub_auth import sub_user # noqa: F401 diff --git a/caterva2/tests/files.py b/caterva2/tests/files.py index f16faeef..c3196b37 100644 --- a/caterva2/tests/files.py +++ b/caterva2/tests/files.py @@ -1,37 +1,17 @@ -import tempfile from pathlib import Path import pytest -try: - from caterva2.services import hdf5root -except ImportError: - hdf5root = None - def get_source_dir(): return Path(__file__).parent.parent.parent def get_examples_dir(): - return get_source_dir() / 'root-example' + return get_source_dir() / "root-example" -@pytest.fixture(scope='session') +@pytest.fixture(scope="session") def examples_dir(): """Directory of read-only example files in Caterva2 source""" return get_examples_dir() - - -def make_examples_hdf5(mkdtemp=lambda: Path(tempfile.mkdtemp())): - if hdf5root is None: - return None - h5fpath = mkdtemp() / 'root-example.h5' - hdf5root.create_example_root(h5fpath) - return h5fpath - - -@pytest.fixture(scope='session') -def examples_hdf5(tmp_path_factory): - """HDF5 file with example datasets in a new temporary directory""" - return make_examples_hdf5(lambda: tmp_path_factory.mktemp('hdf5')) diff --git a/caterva2/tests/services.py b/caterva2/tests/services.py index 5395f0e3..f2ee4bf6 100644 --- a/caterva2/tests/services.py +++ b/caterva2/tests/services.py @@ -1,7 +1,7 @@ """Caterva2 services for tests. -This ensures that Caterva2 broker, publisher and subscriber services are -running before proceeding to tests. It has three modes of operation: +This ensures that Caterva2 subscriber service is running before proceeding to tests. +It has three modes of operation: - Standalone script: when run as a script, it starts the services as children and makes sure that they are available to other local programs. If given an @@ -27,20 +27,6 @@ not tamper with the state directory nor stop the services when tests finish. Usage example: same as above (but on the pytest side). - -- pytest fixture with managed services: if the environment variable - ``CATERVA2_USE_EXTERNAL`` is set to 1, the `services()` fixture uses - external services; otherwise, it takes care of starting the services as - children and making sure that they are available to other local programs. - It also uses the value in `TEST_STATE_DIR` as the directory to store state - in. If the directory exists, it is removed first. Then the directory is - created and populated with the example files from the source distribution. - When tests finish, the services are stopped. - - Usage example:: - - $ cd Caterva2 - $ env CATERVA2_USE_EXTERNAL=1 pytest # state in ``_caterva2_tests`` """ import collections @@ -64,8 +50,7 @@ DEFAULT_STATE_DIR = "_caterva2" TEST_STATE_DIR = DEFAULT_STATE_DIR + "_tests" TEST_DEFAULT_ROOT = "foo" -TEST_CATERVA2_ROOT = TEST_DEFAULT_ROOT -TEST_HDF5_ROOT = "hdf5root" +TEST_CATERVA2_ROOT = "@public" local_port_iter = itertools.count(8100) logger = logging.getLogger("tests") @@ -82,8 +67,6 @@ def get_service_ep(): return get_service_ep -get_bro_ep = service_ep_getter("localhost:8000") -get_pub_ep = service_ep_getter("localhost:8001") get_sub_ep = service_ep_getter("localhost:8002") @@ -105,14 +88,6 @@ def http_service_check(conf, conf_sect, def_host, path): return make_get_http(conf.get(f"{conf_sect}.http", def_host), path) -def bro_check(conf): - return http_service_check(conf, "broker", get_bro_ep(), "/api/roots") - - -def pub_check(id_, conf): - return http_service_check(conf, f"publisher.{id_}", get_pub_ep(), "/api/list") - - def sub_check(conf): return http_service_check(conf, "subscriber", get_sub_ep(), "/api/roots") @@ -120,12 +95,7 @@ def sub_check(conf): TestRoot = collections.namedtuple("TestRoot", ["name", "source"]) -class Services: - def __init__(self): # mostly to appease QA - pass - - -class ManagedServices(Services): +class ManagedServices: def __init__(self, state_dir, reuse_state=True, roots=None, configuration=None): super().__init__() @@ -207,15 +177,6 @@ def _setup(self): def start_all(self): self._setup() - - self._start_proc("broker", check=bro_check(self.configuration)) - for root in self.roots: - self._start_proc( - f"publisher.{root.name}", - root.name, - self._get_data_path(root), - check=pub_check(root.name, self.configuration), - ) self._start_proc("subscriber", check=sub_check(self.configuration)) def stop_all(self): @@ -238,53 +199,14 @@ def get_urlbase(self, service): return f"http://{self.get_endpoint(service)}" -class ExternalServices(Services): - def __init__(self, roots=None, configuration=None): - super().__init__() - self.roots = list(roots) - self.configuration = conf = configuration - - self._checks = checks = {} - checks["broker"] = bro_check(conf) - for root in roots: - checks[f"publisher.{root.name}"] = pub_check(root.name, conf) - checks["subscriber"] = sub_check(conf) - - def start_all(self): - failed = [check.__name__ for check in self._checks.values() if not check()] - if failed: - raise RuntimeError("failed checks for external services: " + " ".join(failed)) - - def stop_all(self): - pass - - def wait_for_all(self): - pass - - def get_endpoint(self, service): - if service not in self._checks: - return None - return self._checks[service].host - - def get_urlbase(self, service): - ep = self.get_endpoint(service) - return f"http://{ep}" if ep else None - - @pytest.fixture(scope="session") -def services(configuration, examples_dir, examples_hdf5): +def services(configuration, examples_dir): # TODO: Consider using a temporary directory to avoid # polluting the current directory with test files # and tests being influenced by the presence of a configuration file. roots = [TestRoot(TEST_CATERVA2_ROOT, examples_dir)] - if examples_hdf5 is not None: - roots.append(TestRoot(TEST_HDF5_ROOT, examples_hdf5)) - srvs = ( - ExternalServices(roots=roots, configuration=configuration) - if os.environ.get("CATERVA2_USE_EXTERNAL", "0") == "1" - else ManagedServices(TEST_STATE_DIR, reuse_state=False, roots=roots, configuration=configuration) - ) + srvs = ManagedServices(TEST_STATE_DIR, reuse_state=False, roots=roots, configuration=configuration) try: srvs.start_all() @@ -313,10 +235,6 @@ def main(defer): from . import conf, files, sub_auth roots = [TestRoot(TEST_DEFAULT_ROOT, files.get_examples_dir())] - hdf5source = files.make_examples_hdf5() - if hdf5source: - defer(lambda: (hdf5source.parent.is_dir() and shutil.rmtree(hdf5source.parent))) - roots.append(TestRoot(TEST_HDF5_ROOT, hdf5source)) if "--help" in sys.argv: rspecs = " ".join(f'"{r.name}={r.source}"' for r in roots) diff --git a/caterva2/tests/test_api.py b/caterva2/tests/test_api.py index aa4548af..cd82f445 100644 --- a/caterva2/tests/test_api.py +++ b/caterva2/tests/test_api.py @@ -20,36 +20,20 @@ from .services import TEST_CATERVA2_ROOT, TEST_STATE_DIR -try: - chdir_ctxt = contextlib.chdir -except AttributeError: # Python < 3.11 - import os - - @contextlib.contextmanager - def chdir_ctxt(path): - cwd = os.getcwd() - os.chdir(path) - yield - os.chdir(cwd) - - -@pytest.fixture -def pub_host(services): - return services.get_endpoint(f"publisher.{TEST_CATERVA2_ROOT}") - @pytest.fixture def fill_public(client, examples_dir): # Manually copy some files to the public area (TEST_STATE_DIR) - fnames = ["README.md", "ds-1d.b2nd", "ds-1d-fields.b2nd", "dir1/ds-2d.b2nd"] + dest_dir = pathlib.Path(TEST_STATE_DIR) / "subscriber/public" + fnames = [str(fname.relative_to(examples_dir)) for fname in examples_dir.rglob("*") if fname.is_file()] for fname in fnames: orig = examples_dir / fname data = orig.read_bytes() - if not fname.endswith(("b2nd", "b2frame")): + if not fname.endswith(("b2nd", "b2frame", "h5")): fname += ".b2" schunk = blosc2.SChunk(data=data) data = schunk.to_cframe() - dest = pathlib.Path(TEST_STATE_DIR) / f"subscriber/public/{fname}" + dest = dest_dir / fname dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(data) # We need a user here in case we want to remove files from @public @@ -64,37 +48,18 @@ def fill_auth(auth_client, fill_public): return fnames, auth_client.get("@public") -def test_subscribe(client, auth_client): - assert client.subscribe(TEST_CATERVA2_ROOT) == "Ok" - assert client.subscribe("@public") == "Ok" - for root in ["@personal", "@shared"]: - if auth_client: - assert auth_client.subscribe(root) == "Ok" - else: - with pytest.raises(Exception) as e_info: - _ = client.subscribe(root) - assert "Unauthorized" in str(e_info) - - -def test_roots(pub_host, client, auth_client): +def test_roots(client, auth_client): client = auth_client if auth_client else client roots = client.get_roots() - assert roots[TEST_CATERVA2_ROOT]["name"] == TEST_CATERVA2_ROOT - assert roots[TEST_CATERVA2_ROOT]["http"] == pub_host assert roots["@public"]["name"] == "@public" - assert roots["@public"]["http"] == "" + + # Special roots (only available when authenticated) if auth_client: - # Special roots (only available when authenticated) assert roots["@personal"]["name"] == "@personal" - assert roots["@personal"]["http"] == "" assert roots["@shared"]["name"] == "@shared" - assert roots["@shared"]["http"] == "" def test_get_root(client, auth_client): - myroot = client.get(TEST_CATERVA2_ROOT) - assert myroot.name == TEST_CATERVA2_ROOT - assert myroot.urlbase == client.urlbase mypublic = client.get("@public") assert mypublic.name == "@public" assert mypublic.urlbase == client.urlbase @@ -107,13 +72,13 @@ def test_get_root(client, auth_client): assert myshared.urlbase == auth_client.urlbase -def test_get_file(client): - myfile = client.get(TEST_CATERVA2_ROOT + "/README.md") +def test_get_file(client, fill_public): + myfile = client.get("@public/README.md") assert myfile.name == "README.md" -def test_get_dataset(client): - myds = client.get(TEST_CATERVA2_ROOT + "/ds-1d.b2nd") +def test_get_dataset(client, fill_public): + myds = client.get("@public/ds-1d.b2nd") assert myds.name == "ds-1d.b2nd" assert isinstance(myds, cat2.Dataset) assert myds.shape == (1000,) @@ -124,10 +89,6 @@ def test_get_dataset(client): def test_list(client, auth_client, examples_dir): - myroot = client.get(TEST_CATERVA2_ROOT) - example = examples_dir - files = {str(f.relative_to(str(example))) for f in example.rglob("*") if f.is_file()} - assert set(myroot.file_list) == files if auth_client: mypersonal = auth_client.get("@personal") # In previous tests we have created some files in the personal area @@ -141,31 +102,13 @@ def test_list_public(client, fill_public): assert set(mypublic.file_list) == set(fnames) # Test toplevel list flist = client.get_list("@public") - assert len(flist) == 4 - for fname in flist: - assert fname in fnames + assert set(flist) == set(fnames) # Test directory list - flist = client.get_list("@public/dir1") - assert len(flist) == 1 - for fname in flist: - assert fname == "ds-2d.b2nd" + assert client.get_list("@public/dir1") == ["ds-2d.b2nd", "ds-3d.b2nd"] # Test directory list with trailing slash - flist = client.get_list("@public/dir1/") - assert len(flist) == 1 - for fname in flist: - assert fname == "ds-2d.b2nd" + assert client.get_list("@public/dir1/") == ["ds-2d.b2nd", "ds-3d.b2nd"] # Test single dataset list - flist = client.get_list("@public/dir1/ds-2d.b2nd") - assert len(flist) == 1 - for fname in flist: - assert fname == "ds-2d.b2nd" - - -def test_file(client): - myroot = client.get(TEST_CATERVA2_ROOT) - file = myroot["README.md"] - assert file.name == "README.md" - assert file.urlbase == client.urlbase + assert client.get_list("@public/dir1/ds-2d.b2nd") == ["ds-2d.b2nd"] def test_file_public(client, fill_public): @@ -179,9 +122,9 @@ def test_file_public(client, fill_public): def test_dataset_info(client, fill_public): fnames, mypublic = fill_public for fname in fnames: - if type(mypublic[fname]) is cat2.Dataset: # Files cannot be expected to have attributes - info = client.get_info("@public/" + fname) + if fname.endswith(".b2nd"): data = mypublic[fname] + info = client.get_info("@public/" + fname) assert data.dtype == info["dtype"] assert data.shape == tuple(info["shape"]) assert data.blocks == tuple(info["blocks"]) @@ -386,7 +329,7 @@ def test_append(auth_client, fields, fill_auth, examples_dir): "slice_", [1, slice(None, 1), slice(0, 10), slice(10, 20), slice(None), slice(10, 20, 1)], ) -def test_dataset_getitem_fetch(slice_, examples_dir, client): +def test_dataset_getitem_fetch(slice_, examples_dir, client, fill_public): myroot = client.get(TEST_CATERVA2_ROOT) ds = myroot["ds-hello.b2frame"] assert ds.name == "ds-hello.b2frame" @@ -538,14 +481,14 @@ def test_getitem_client_nd(slice_, name, examples_dir, client): def test_download_b2nd(name, examples_dir, tmp_path, client, auth_client): myroot = client.get(TEST_CATERVA2_ROOT) ds = myroot[name] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path # Data contents example = examples_dir / name a = blosc2.open(example) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): b = blosc2.open(path) np.testing.assert_array_equal(a[:], b[:]) @@ -560,14 +503,14 @@ def test_download_b2nd(name, examples_dir, tmp_path, client, auth_client): def test_download_b2frame(examples_dir, tmp_path, client, auth_client): myroot = client.get(TEST_CATERVA2_ROOT) ds = myroot["ds-hello.b2frame"] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path # Data contents example = examples_dir / ds.name a = blosc2.open(example) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): b = blosc2.open(path) assert a[:] == b[:] @@ -593,7 +536,7 @@ def test_download_localpath(fnames, examples_dir, tmp_path, client): myroot = client.get(TEST_CATERVA2_ROOT) name, localpath = fnames ds = myroot[name] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): if localpath.endswith("/"): # Create a directory in localpath localpath2 = pathlib.Path(localpath) @@ -606,7 +549,7 @@ def test_download_localpath(fnames, examples_dir, tmp_path, client): # Data contents example = examples_dir / name a = blosc2.open(example) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): b = blosc2.open(path) np.testing.assert_array_equal(a[:], b[:]) @@ -614,14 +557,14 @@ def test_download_localpath(fnames, examples_dir, tmp_path, client): def test_download_regular_file(examples_dir, tmp_path, client, auth_client): myroot = client.get(TEST_CATERVA2_ROOT) ds = myroot["README.md"] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path # Data contents example = examples_dir / ds.name a = open(example).read() - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): b = open(path).read() assert a[:] == b[:] @@ -638,9 +581,13 @@ def test_download_regular_file(examples_dir, tmp_path, client, auth_client): def test_download_public_file(examples_dir, fill_public, tmp_path): fnames, mypublic = fill_public for fname in fnames: + # TODO fetch (and download) of HDF5 files is not supported (gives a 500 error) + if fname.endswith(".h5"): + continue + # Download the file ds = mypublic[fname] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path # Check data contents @@ -673,7 +620,7 @@ def test_upload(fnames, remove, root, examples_dir, tmp_path, auth_client): remote_root = auth_client.get(root) myroot = auth_client.get(TEST_CATERVA2_ROOT) ds = myroot[localpath] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path # Check whether path exists and is a file @@ -706,7 +653,7 @@ def test_upload_public_unauthorized(client, auth_client, examples_dir, tmp_path) remote_root = client.get("@public") myroot = client.get(TEST_CATERVA2_ROOT) ds = myroot["README.md"] - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): path = ds.download() assert path == ds.path with pytest.raises(Exception) as e_info: @@ -741,7 +688,6 @@ def test_lazyexpr(auth_client): operands = {opnm: oppt} lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) opinfo = auth_client.get_info(oppt) lxpath = auth_client.lazyexpr(lxname, expression, operands) assert lxpath == pathlib.Path(f"@personal/{lxname}.b2nd") @@ -779,7 +725,7 @@ def test_lazyexpr2(expression, examples_dir, tmp_path, auth_client): ds_a = f"{remote_dir}/3d-blosc2-a.b2nd" ds_b = f"{remote_dir}/3d-blosc2-b.b2nd" - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): os.makedirs(remote_dir, exist_ok=True) a = np.linspace(-1, 2, 1000).reshape(10, 10, 10) blosc2.asarray(a, urlpath=ds_a, chunks=(5, 10, 10)) @@ -803,7 +749,7 @@ def test_lazyexpr2(expression, examples_dir, tmp_path, auth_client): np.testing.assert_allclose(result[:], nresult) -def test_lazyexpr_getchunk(auth_client): +def test_lazyexpr_getchunk(auth_client, fill_public): if not auth_client: pytest.skip("authentication support needed") @@ -813,7 +759,6 @@ def test_lazyexpr_getchunk(auth_client): operands = {opnm: oppt} lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) lxpath = auth_client.lazyexpr(lxname, expression, operands) assert lxpath == pathlib.Path(f"@personal/{lxname}.b2nd") @@ -838,7 +783,6 @@ def test_lazyexpr_fields(auth_client): pytest.skip("authentication support needed") oppt = f"{TEST_CATERVA2_ROOT}/ds-1d-fields.b2nd" - auth_client.subscribe(TEST_CATERVA2_ROOT) # Test a field arr = auth_client.get(oppt) @@ -902,7 +846,6 @@ def test_expr_from_expr(auth_client): operands = {opnm: oppt} lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) opinfo = auth_client.get_info(oppt) lxpath = auth_client.lazyexpr(lxname, expression, operands) assert lxpath == pathlib.Path(f"@personal/{lxname}.b2nd") @@ -939,7 +882,6 @@ def test_expr_no_operand(auth_client): expression = "linspace(0, 10, num=50)" lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) lxpath = auth_client.lazyexpr(lxname, expression) assert lxpath == pathlib.Path(f"@personal/{lxname}.b2nd") c = auth_client.get(lxpath) @@ -952,7 +894,6 @@ def test_expr_no_operand(auth_client): expression = "ds + linspace(0, 10, num=50)" lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) with pytest.raises(Exception) as e_info: lxpath = auth_client.lazyexpr(lxname, expression) @@ -964,8 +905,6 @@ def test_expr_force_compute(auth_client): expression = "linspace(0, 10, num=50)" lxname = "my_expr" - auth_client.subscribe(TEST_CATERVA2_ROOT) - # Uncomputed lazyexpr is a blosc2 lazyexpr lxpath = auth_client.lazyexpr(lxname, expression, compute=False) assert lxpath == pathlib.Path(f"@personal/{lxname}.b2nd") @@ -1114,7 +1053,6 @@ def test_client_timeout(auth_client): if not auth_client: pytest.skip("authentication support needed") - auth_client.subscribe(TEST_CATERVA2_ROOT) lxpath = auth_client.lazyexpr("expr", "linspace(0, 100, 1000_0000)", compute=True) assert lxpath == pathlib.Path("@personal/expr.b2nd") auth_client.timeout = 0.0001 diff --git a/caterva2/tests/test_cli.py b/caterva2/tests/test_cli.py index c0c87373..2d43b1a2 100644 --- a/caterva2/tests/test_cli.py +++ b/caterva2/tests/test_cli.py @@ -17,54 +17,30 @@ from .services import TEST_CATERVA2_ROOT -@pytest.fixture -def pub_host(services): - return services.get_endpoint(f'publisher.{TEST_CATERVA2_ROOT}') - - @pytest.fixture def sub_urlbase(services): - return services.get_urlbase('subscriber') + return services.get_urlbase("subscriber") def cli(cargs, binary=False, sub_user=None) -> str or dict: - cli_path = 'caterva2.clients.cli' - args = [sys.executable, '-m' + str(cli_path)] + cli_path = "caterva2.clients.cli" + args = [sys.executable, "-m" + str(cli_path)] if sub_user: - args += ['--username', sub_user.username, - '--password', sub_user.password] + args += ["--username", sub_user.username, "--password", sub_user.password] args += cargs if not binary: - args += ['--json'] + args += ["--json"] ret = subprocess.run(args, capture_output=True, text=True) assert ret.returncode == 0 out = ret.stdout return out if binary else json.loads(out) -def test_roots(pub_host, sub_user): - roots = cli(['roots'], sub_user=sub_user) - assert roots[TEST_CATERVA2_ROOT]['name'] == TEST_CATERVA2_ROOT - assert roots[TEST_CATERVA2_ROOT]['http'] == pub_host +def test_roots(sub_user): + roots = cli(["roots"], sub_user=sub_user) + assert roots[TEST_CATERVA2_ROOT]["name"] == TEST_CATERVA2_ROOT def test_url(sub_urlbase, sub_user): - out = cli(['url', f'{TEST_CATERVA2_ROOT}/ds-1d.b2nd'], sub_user=sub_user) - assert out == f'{sub_urlbase}/api/fetch/{TEST_CATERVA2_ROOT}/ds-1d.b2nd' - - -def test_subscribe(sub_user): - # Subscribe once - out = cli(['subscribe', TEST_CATERVA2_ROOT], sub_user=sub_user) - assert out == 'Ok' - - # Subscribe again, should be a noop - out = cli(['subscribe', TEST_CATERVA2_ROOT], sub_user=sub_user) - assert out == 'Ok' - - # Show - a = cli(['show', f'{TEST_CATERVA2_ROOT}/ds-1d.b2nd'], - binary=True, sub_user=sub_user) - b = cli(['show', f'{TEST_CATERVA2_ROOT}/ds-1d.b2nd'], - binary=True, sub_user=sub_user) - assert a == b + out = cli(["url", f"{TEST_CATERVA2_ROOT}/ds-1d.b2nd"], sub_user=sub_user) + assert out == f"{sub_urlbase}/api/fetch/{TEST_CATERVA2_ROOT}/ds-1d.b2nd" diff --git a/caterva2/tests/test_hdf5_proxy.py b/caterva2/tests/test_hdf5_proxy.py index f8bae73d..21cc3dab 100644 --- a/caterva2/tests/test_hdf5_proxy.py +++ b/caterva2/tests/test_hdf5_proxy.py @@ -15,20 +15,83 @@ import numexpr as ne import numpy as np import pytest +from hdf5plugin import Blosc2 as B2Comp -hdf5root = pytest.importorskip("caterva2.services.hdf5root", reason="HDF5 support not present") -try: - chdir_ctxt = contextlib.chdir -except AttributeError: # Python < 3.11 - import os +def create_example_root(path): + """Create an example HDF5 file to be used as a root.""" - @contextlib.contextmanager - def chdir_ctxt(path): - cwd = os.getcwd() - os.chdir(path) - yield - os.chdir(cwd) + with h5py.File(path, "x") as h5f: + h5f.create_dataset("/scalar", data=123.456) + h5f.create_dataset("/string", data=np.bytes_("Hello world!")) + + a = np.arange(100, dtype="uint8") + h5f.create_dataset("/arrays/1d-raw", data=a) + + h5f["/arrays/soft-link"] = h5py.SoftLink("/arrays/1d-raw") + + a = np.array([b"foobar"] * 100) + h5f.create_dataset("/arrays/1ds-blosc2", data=a, chunks=(50,), **B2Comp()) + + a = np.arange(100, dtype="complex128").reshape(10, 10) + a = a + a * 1j + h5f.create_dataset("/arrays/2d-nochunks", data=a, chunks=None) + + a = np.arange(100, dtype="complex128").reshape(10, 10) + a = a + a * 1j + h5f.create_dataset("/arrays/2d-gzip", data=a, chunks=(4, 4), compression="gzip") + + a = np.arange(1000, dtype="uint8").reshape(10, 10, 10) + h5f.create_dataset( + "/arrays/3d-blosc2", + data=a, + chunks=(4, 10, 10), + **B2Comp(cname="lz4", clevel=7, filters=B2Comp.BITSHUFFLE), + ) + + a = np.linspace(-1, 2, 1000).reshape(10, 10, 10) + h5f.create_dataset( + "/arrays/3d-blosc2-a", + data=a, + chunks=(4, 10, 10), + **B2Comp(cname="lz4", clevel=7, filters=B2Comp.BITSHUFFLE), + ) + + a = np.linspace(-1, 2, 1000).reshape(10, 10, 10) + h5f.create_dataset( + "/arrays/3d-blosc2-b", + data=a, + chunks=(2, 5, 10), + **B2Comp(cname="blosclz", clevel=7, filters=B2Comp.SHUFFLE), + ) + + h5f.create_dataset("/arrays/array-dtype", dtype=np.dtype(("float64", (4,))), shape=(10,)) + + ds = h5f.create_dataset("/attrs", data=0) + a = np.arange(4, dtype="uint8").reshape(2, 2) + for k, v in { + "Int": 42, + "IntT": np.int16(42), + "Bin": b"foo", + "BinT": np.bytes_(b"foo"), + "Str": "bar", # StrT=np.str_("bar"), + "Arr": a.tolist(), + "ArrT": a, + "NilBin": h5py.Empty("|S4"), + # NilStr=h5py.Empty('|U4'), + "NilInt": h5py.Empty("uint8"), + }.items(): + ds.attrs[k] = v + + h5f.create_dataset("/arrays/empty", data=h5py.Empty("float64")) + + h5f.create_dataset("/arrays/compound-dtype", dtype=np.dtype("uint8,float64"), shape=(10,)) + + a = np.arange(1, dtype="uint8").reshape((1,) * 23) + h5f.create_dataset("/unsupported/too-many-dimensions", data=a) + + # TODO: This could be supported by mapping the vlstring dataset to an NDArray with shape=() and dtype="u1" + h5f.create_dataset("/unsupported/vlstring", data="Hello world!") def get_all_datasets(f, prefix=""): @@ -50,7 +113,7 @@ def get_all_datasets(f, prefix=""): ("ex-noattr.h5", None), ("ex-noattr.h5", "ex-noattr2.h5"), # upload with remote name ("root-example.h5", None), - (None, None), # use hdf5root.create_example_root(h5fpath) + (None, None), # use create_example_root(h5fpath) ], ) @pytest.mark.parametrize("root", ["@personal", "@shared", "@public"]) @@ -62,12 +125,12 @@ def test_unfold(fnames, remove, root, examples_dir, tmp_path, auth_client): # First, choose an HDF5 dataset localpath, remotepath = fnames remote_root = auth_client.get(root) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): if localpath is None: # Create a temporary HDF5 file localpath = "create-example-root.h5" path = tmp_path / localpath - hdf5root.create_example_root(path) + create_example_root(path) else: path = examples_dir / localpath assert path.is_file() @@ -130,7 +193,7 @@ def create_and_unfold_hdf5(tmp_path, remote_root, create_file=True, localpath="c if create_file: # Create a temporary HDF5 file - hdf5root.create_example_root(hdf5_path) + create_example_root(hdf5_path) # Upload the file to the remote root remote_ds = remote_root.upload(localpath) @@ -156,7 +219,7 @@ def test_unfold_download(examples_dir, tmp_path, auth_client): root = pathlib.Path("@shared") remote_root = auth_client.get(root) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): hdf5_path, remote_dir, file_list = create_and_unfold_hdf5(tmp_path, remote_root) h5f = h5py.File(hdf5_path, "r") for file_ in file_list: @@ -197,7 +260,7 @@ def test_unfold_fetch(fetch_or_slice, examples_dir, tmp_path, auth_client): root = pathlib.Path("@shared") remote_root = auth_client.get(root) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): hdf5_path, remote_dir, file_list = create_and_unfold_hdf5(tmp_path, remote_root) h5f = h5py.File(hdf5_path, "r") for file_ in file_list: @@ -249,7 +312,7 @@ def test_expression(expression, examples_dir, tmp_path, auth_client): ds_b = "arrays/3d-blosc2-b" root = pathlib.Path("@shared") remote_root = auth_client.get(root) - with chdir_ctxt(tmp_path): + with contextlib.chdir(tmp_path): hdf5_path, remote_dir, file_list = create_and_unfold_hdf5(tmp_path, remote_root) h5f = h5py.File(hdf5_path, "r") remote_a = remote_dir / (ds_a + ".b2nd") diff --git a/caterva2/tests/test_hdf5root.py b/caterva2/tests/test_hdf5root.py deleted file mode 100644 index 40fbb689..00000000 --- a/caterva2/tests/test_hdf5root.py +++ /dev/null @@ -1,122 +0,0 @@ -############################################################################### -# Caterva2 - On demand access to remote Blosc2 data repositories -# -# Copyright (c) 2023 ironArray SLU -# https://www.blosc.org -# License: GNU Affero General Public License v3.0 -# See LICENSE.txt for details about copyright and rights to use. -############################################################################### - -import numpy as np -import pytest - -from .services import TEST_HDF5_ROOT - -hdf5root = pytest.importorskip("caterva2.services.hdf5root", reason="HDF5 support not present") - - -@pytest.fixture -def sub_urlbase(services): - return services.get_urlbase("subscriber") - - -@pytest.fixture -def api_root(client): - return client.get(TEST_HDF5_ROOT) - - -def test_not_unsupported(api_root): - for node in api_root.file_list: - assert not node.startswith("unsupported/") - - -def test_ds_name_ext(api_root): - for node in api_root.file_list: - node.endswith(".b2nd") # no other conversions supported yet - - -def test_scalar(api_root): - ds = api_root["scalar.b2nd"] - v = ds[()] - assert v.dtype.kind == "f" - assert v == pytest.approx(123.456) - - -def test_string(api_root): - ds = api_root["string.b2nd"] - v = ds[()] - assert v.dtype.kind == "S" - assert v == b"Hello world!" - - -def test_1d_raw(api_root): - ds = api_root["arrays/1d-raw.b2nd"] - v = ds[()] - assert v.dtype.kind == "u" - np.testing.assert_array_equal(v, np.arange(100, dtype="uint8")) - - -@pytest.mark.skip("Softlink not supported in this context yet") -def test_softlink(api_root): - ds = api_root["arrays/soft-link.b2nd"] # TODO: this fails here - v = ds[()] - assert v.dtype.kind == "u" - np.testing.assert_array_equal(v, np.arange(100, dtype="uint8")) - - -def test_nonchunked(api_root): - ds = api_root["arrays/2d-nochunks.b2nd"] - ds_chunks = ds.meta["chunks"] - assert ds_chunks is not None - assert len(ds_chunks) == 2 - v = ds[:] - a = np.arange(100, dtype="complex128").reshape(10, 10) - a = a + a * 1j - np.testing.assert_array_equal(v, a) - - -def test_chunked(api_root): - ds = api_root["arrays/2d-gzip.b2nd"] - ds_chunks = tuple(ds.meta["chunks"]) - assert ds_chunks == (4, 4) # chunk shape is kept - v = ds[:] - a = np.arange(100, dtype="complex128").reshape(10, 10) - a = a + a * 1j - np.testing.assert_array_equal(v, a) - - -def test_blosc2(api_root): - ds = api_root["arrays/3d-blosc2.b2nd"] - ds_chunks = tuple(ds.meta["chunks"]) - assert ds_chunks == (4, 10, 10) # chunk shape is kept - # TODO: compression parameters - # cparams = ds.meta['schunk']['cparams'] - # assert cparams['codec'] == blosc2.Codec.LZ4.value - # assert cparams['filters'] == [0, 0, 0, 0, 0, - # blosc2.Filter.BITSHUFFLE.value] - v = ds[:] - a = np.arange(1000, dtype="uint8").reshape(10, 10, 10) - np.testing.assert_array_equal(v, a) - - -@pytest.mark.parametrize("slice_", [slice(None), 1, slice(2, 6), (slice(None, 6), slice(5, 8), slice(6))]) -def test_slicing(api_root, slice_): - ds = api_root["arrays/3d-blosc2.b2nd"] - v = ds[:] - a = np.arange(1000, dtype="uint8").reshape(10, 10, 10) - np.testing.assert_array_equal(v[slice_], a[slice_]) - - -def test_vlmeta(api_root): - ds = api_root["attrs.b2nd"] - assert len(ds.vlmeta) == 9 - m = ds.vlmeta - assert m["Int"] == m["IntT"] == 42 - # TODO: consistent conversion of strings - # assert m['Bin'] == m['BinT'] == b'foo' - # assert m['Str'] == m['StrT'] == 'bar' - assert m["Arr"] == m["ArrT"] == [[0, 1], [2, 3]] - # TODO: consistent conversion of strings - # assert m['NilBin'] == b'' - # assert m['NilStr'] == '' - assert m["NilInt"] is None diff --git a/caterva2/utils.py b/caterva2/utils.py index 124c9c89..af2838c1 100644 --- a/caterva2/utils.py +++ b/caterva2/utils.py @@ -8,29 +8,11 @@ ############################################################################### import argparse -import contextlib import datetime import logging import os import pathlib - -try: - import tomllib as toml -except ImportError: - import tomli as toml - -# -# Context managers -# - - -@contextlib.contextmanager -def log_exception(logger, message): - try: - yield - except Exception: - logger.exception(message) - +import tomllib as toml # # Datetime related @@ -96,11 +78,9 @@ def __init__(self, string): self.uds = string -def get_parser(loglevel="warning", statedir=None, id=None, http=None, url=None, broker=None): +def get_parser(loglevel="warning", statedir=None, id=None, http=None): parser = argparse.ArgumentParser() _add_preliminary_args(parser, id=id) # just for help purposes - if broker is not None: - parser.add_argument("--broker", default=broker, type=Socket, help="socket address of the broker") if http is not None: parser.add_argument( "--http", default=http, type=Socket, help="Listen to given hostname:port or unix socket" @@ -125,8 +105,6 @@ def run_parser(parser): # Configuration file # -conf_file_name = "caterva2.toml" - class Conf: def __init__(self, conf, prefix=None, id=None): @@ -158,7 +136,7 @@ def get(self, key, default=None): def _add_preliminary_args(parser, id=None): parser.add_argument( "--conf", - default=conf_file_name, + default="caterva2.toml", type=pathlib.Path, help=("path to alternative configuration file " "(may not exist)"), ) @@ -168,12 +146,6 @@ def _add_preliminary_args(parser, id=None): ) -def _parse_preliminary_args(allow_id): - parser = argparse.ArgumentParser(add_help=False) - _add_preliminary_args(parser, id="" if allow_id else None) - return parser.parse_known_args()[0] - - def get_conf(prefix=None, allow_id=False): """Get settings from the configuration file, if existing. @@ -189,7 +161,9 @@ def get_conf(prefix=None, allow_id=False): For instance, with ``conf = get_conf('foo')`` and ``--id=bar``, ``conf.get('.item')`` is equivalent to ``conf.get('foo.bar.item')``. """ - opts = _parse_preliminary_args(allow_id) + parser = argparse.ArgumentParser(add_help=False) + _add_preliminary_args(parser, id="" if allow_id else None) + opts = parser.parse_known_args()[0] if allow_id and opts.id and any(p in opts.id for p in [os.curdir, os.pardir, os.sep]): raise ValueError("invalid identifier", opts.id) diff --git a/doc/tutorials/independent-services.md b/doc/tutorials/independent-services.md index 6e687c44..934d4852 100644 --- a/doc/tutorials/independent-services.md +++ b/doc/tutorials/independent-services.md @@ -172,20 +172,22 @@ cat2cli roots When using the programmatic API, you need to provide the subscriber address explicitly: ```python -roots = caterva2.get_roots(urlbase='http://sub.edu.example.org:3126') -foo = caterva2.Root('foo', urlbase='http://sub.edu.example.org:3126') +roots = caterva2.get_roots(urlbase="http://sub.edu.example.org:3126") +foo = caterva2.Root("foo", urlbase="http://sub.edu.example.org:3126") ``` Since parsing TOML is very easy with Python, your API client may just access the needed configuration like this: ``` python -from tomllib import load as toml_load # "from tomli" on Python < 3.11 -with open('caterva2.toml', 'rb') as conf_file: +from tomllib import load as toml_load + +with open("caterva2.toml", "rb") as conf_file: conf = toml_load(conf_file) -#user_auth = dict(username=conf['client]['username'], +# user_auth = dict(username=conf['client]['username'], # password=conf['client]['password']) -foo = caterva2.Root('foo', - urlbase=conf['subscriber']['url'], - #user_auth=user_auth, +foo = caterva2.Root( + "foo", + urlbase=conf["subscriber"]["url"], + # user_auth=user_auth, ) ``` diff --git a/pyproject.toml b/pyproject.toml index 1dbb28c3..c53eb7b8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,7 +21,7 @@ authors = [ {name = "ironArray SLU", email = "contact@ironarray.io"}, ] dynamic = ["version"] -keywords = ["pubsub", "blosc2"] +keywords = ["blosc2"] license = {text = "GNU Affero General Public License version 3"} classifiers = [ "Programming Language :: Python :: 3", @@ -37,7 +37,6 @@ classifiers = [ dependencies = [ "blosc2>=3.2.0", "httpx", - "tomli>=2;python_version<\"3.11\"", ] [tool.hatch.version] @@ -46,7 +45,6 @@ path = "caterva2/__init__.py" [project.optional-dependencies] base-services = [ "fastapi>=0.109", - "fastapi_websocket_pubsub", "pydantic>=2", "safer", "uvicorn", @@ -109,8 +107,6 @@ only-include = ["caterva2", "root-example"] Home = "https://github.com/ironArray/Caterva2" [project.scripts] -cat2bro = "caterva2.services.bro:main" -cat2pub = "caterva2.services.pub:main" cat2sub = "caterva2.services.sub:main" cat2agent = "caterva2.clients.agent:main" cat2cli = "caterva2.clients.cli:main"