chris.client.admin

  1import io
  2import json
  3from typing import Iterable
  4
  5import aiohttp
  6from async_property import async_cached_property
  7from serde import from_dict
  8
  9from chris.client.authed import AuthenticatedClient
 10from chris.link import http
 11from chris.link.collection_client import CollectionJsonApiClient
 12from chris.link.linked import deserialize_linked
 13from chris.models.collection_links import AdminCollectionLinks, AdminApiCollectionLinks
 14from chris.models.logged_in import Plugin
 15from chris.models.public import ComputeResource
 16from chris.models.types import PluginUrl, ComputeResourceName, PfconUrl
 17from chris.util.errors import raise_for_status
 18
 19
 20class _AdminApiClient(CollectionJsonApiClient[AdminApiCollectionLinks]):
 21    """
 22    A client to `/chris-admin/api/v1/`
 23    """
 24
 25    @http.post("compute_resources")
 26    async def create_compute_resource(self, **kwargs) -> ComputeResource:
 27        ...
 28
 29
 30class ChrisAdminClient(AuthenticatedClient[AdminCollectionLinks, "ChrisAdminClient"]):
 31    """
 32    A client who has access to `/chris-admin/`. Admins can register new plugins and
 33    add new compute resources.
 34    """
 35
 36    @http.post("admin")
 37    async def _register_plugin_from_store_raw(
 38        self, plugin_store_url: str, compute_names: str
 39    ) -> Plugin:
 40        ...
 41
 42    async def register_plugin_from_store(
 43        self, plugin_store_url: PluginUrl, compute_names: Iterable[ComputeResourceName]
 44    ) -> Plugin:
 45        """
 46        Register a plugin from a ChRIS Store.
 47        """
 48        return await self._register_plugin_from_store_raw(
 49            plugin_store_url=plugin_store_url, compute_names=",".join(compute_names)
 50        )
 51
 52    async def add_plugin(
 53        self,
 54        plugin_description: str | dict,
 55        compute_resources: str
 56        | ComputeResource
 57        | Iterable[ComputeResource | ComputeResourceName],
 58        fname: str = "Plugin.json",
 59    ) -> Plugin:
 60        """
 61        Add a plugin to *CUBE*.
 62
 63        Examples
 64        --------
 65
 66        ```python
 67        cmd = ['docker', 'run', '--rm', 'fnndsc/pl-mri-preview', 'chris_plugin_info']
 68        output = subprocess.check_output(cmd, text=True)
 69        desc = json.loads(output)
 70        desc['name'] = 'pl-mri-preview'
 71        desc['public_repo'] = 'https://github.com/FNNDSC/pl-mri-preview'
 72        desc['dock_image'] = 'fnndsc/pl-mri-preview'
 73
 74        await chris_admin.add_plugin(plugin_description=desc, compute_resources='host')
 75        ```
 76
 77        The example above is just for show. It's not a good example for several reasons:
 78
 79        - Calls blocking function `subprocess.check_output` in asynchronous context
 80        - It is preferred to use a versioned string for `dock_image`
 81        - `host` compute environment is not guaranteed to exist. Instead, you could
 82          call `chris.client.authed.AuthenticatedClient.search_compute_resources`
 83          or `chris.client.authed.AuthenticatedClient.get_all_compute_resources`:
 84
 85        ```python
 86        all_computes = await chris_admin.get_all_compute_resources()
 87        await chris_admin.add_plugin(plugin_description=desc, compute_resources=all_computes)
 88        ```
 89
 90        Parameters
 91        ----------
 92        plugin_description: str | dict
 93            JSON description of a plugin.
 94            [spec](https://github.com/FNNDSC/CHRIS_docs/blob/5078aaf934bdbe313e85367f88aff7c14730a1d4/specs/ChRIS_Plugins.adoc#descriptor_file)
 95        compute_resources
 96            Compute resources to register the plugin to. Value can be either a comma-separated `str` of names,
 97            a `chris.models.public.ComputeResource`, a sequence of `chris.models.public.ComputeResource`,
 98            or a sequence of compute resource names as `str`.
 99        fname: str
100            File name to send along in the multi-part POST request. Not important.
101        """
102        compute_names = _serialize_crs(compute_resources)
103        if not isinstance(plugin_description, str):
104            plugin_description = json.dumps(plugin_description)
105        data = aiohttp.FormData()
106        data.add_field("fname", io.StringIO(plugin_description), filename=fname)
107        data.add_field("compute_names", compute_names)
108        async with self.s.post(self.collection_links.admin, data=data) as res:
109            await raise_for_status(res)
110            return deserialize_linked(self, Plugin, await res.json())
111
112    async def create_compute_resource(
113        self,
114        name: str | ComputeResourceName,
115        compute_url: str | PfconUrl,
116        compute_user: str,
117        compute_password: str,
118        description: str = None,
119        compute_auth_url: str = None,
120        compute_auth_token: str = None,
121        max_job_exec_seconds: str = None,
122    ) -> ComputeResource:
123        """
124        Define a new compute resource.
125        """
126        return await (await self._admin).create_compute_resource(
127            name=name,
128            compute_url=compute_url,
129            compute_user=compute_user,
130            compute_password=compute_password,
131            description=description,
132            compute_auth_url=compute_auth_url,
133            compute_auth_token=compute_auth_token,
134            max_job_exec_seconds=max_job_exec_seconds,
135        )
136
137    @async_cached_property
138    async def _admin(self) -> _AdminApiClient:
139        """
140        Get a (sub-)client for `/chris-admin/api/v1/`
141        """
142        res = await self.s.get(self.collection_links.admin)
143        body = await res.json()
144        links = from_dict(AdminApiCollectionLinks, body["collection_links"])
145        return _AdminApiClient(
146            url=self.collection_links.admin,
147            s=self.s,
148            collection_links=links,
149            max_search_requests=self.max_search_requests,
150        )
151
152
153def _serialize_crs(
154    c: str | ComputeResource | Iterable[ComputeResource | ComputeResourceName],
155) -> str:
156    if isinstance(c, str):
157        return c
158    if isinstance(c, ComputeResource):
159        return c.name
160    if not isinstance(c, Iterable):
161        raise TypeError("compute_resources must be str or Iterable")
162    return ",".join(map(_serialize_cr, c))
163
164
165def _serialize_cr(c: str | ComputeResource):
166    if isinstance(c, ComputeResource):
167        return c.name
168    return str(c)