chris

13class AnonChrisClient(BaseChrisClient[AnonymousCollectionLinks, "AnonChrisClient"]):
14    """
15    An anonymous ChRIS client. It has access to read-only GET resources,
16    such as being able to search for plugins.
17    """
18
19    @classmethod
20    async def from_url(
21        cls,
22        url: str,
23        max_search_requests: int = 100,
24        connector: Optional[aiohttp.BaseConnector] = None,
25        connector_owner: bool = True,
26    ) -> "AnonChrisClient":
27        """
28        Create an anonymous client.
29
30        See `chris.client.base.BaseChrisClient.new` for parameter documentation.
31        """
32        return await cls.new(
33            url=url,
34            max_search_requests=max_search_requests,
35            connector=connector,
36            connector_owner=connector_owner,
37        )
38
39    @http.search("plugins")
40    def search_plugins(self, **query) -> Search[PublicPlugin]:
41        """
42        Search for plugins.
43
44        Since this client is not logged in, you cannot create plugin instances using this client.
45        """
46        ...

15class ChrisClient(AuthenticatedClient[CollectionLinks, "ChrisClient"]):
16    """
17    A normal user *ChRIS* client, who may upload files and create plugin instances.
18    """
19
20    @classmethod
21    async def create_user(
22        cls,
23        url: ChrisURL | str,
24        username: Username | str,
25        password: Password | str,
26        email: str,
27        session: Optional[aiohttp.ClientSession],
28    ) -> UserData:
29        payload = {
30            "template": {
31                "data": [
32                    {"name": "email", "value": email},
33                    {"name": "username", "value": username},
34                    {"name": "password", "value": password},
35                ]
36            }
37        }
38        headers = {
39            "Content-Type": "application/vnd.collection+json",
40            "Accept": "application/json",
41        }
42        async with _optional_session(session) as session:
43            res = await session.post(url + "users/", json=payload, headers=headers)
44            await raise_for_status(res)
45            return from_json(UserData, await res.text())

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

 41@dataclass
 42class Search(Generic[T], AsyncIterable[T]):
 43    """
 44    Abstraction over paginated collection responses from *CUBE*.
 45    `Search` objects are returned by methods for search endpoints of the *CUBE* API.
 46    It is an [asynchronous iterable](https://docs.python.org/3/glossary.html#term-asynchronous-iterable)
 47    which produces items from responses that return multiple results.
 48    HTTP requests are fired as-neede, they happen in the background during iteration.
 49    No request is made before the first time a `Search` object is called.
 50
 51    .. note:: Pagination is handled internally and automatically.
 52             The query parameters `limit` and `offset` can be explicitly given, but they shouldn't.
 53
 54    Examples
 55    --------
 56
 57    Use an `async for` loop to print the name of every feed:
 58
 59    ```python
 60    all_feeds = chris.search_feeds()  # returns a Search[Feed]
 61    async for feed in all_feeds:
 62        print(feed.name)
 63    ```
 64    """
 65
 66    base_url: str
 67    params: dict[str, Any]
 68    client: Linked
 69    Item: Type[T]
 70    max_requests: int = 100
 71
 72    def __aiter__(self) -> AsyncIterator[T]:
 73        return self._paginate(self.url)
 74
 75    async def first(self) -> Optional[T]:
 76        """
 77        Get the first item.
 78
 79        See also
 80        --------
 81        `get_only` : similar use, but more strict
 82        """
 83        return await anext(self._first_aiter(), None)
 84
 85    async def get_only(self, allow_multiple=False) -> T:
 86        """
 87        Get the *only* item from a search with one result.
 88
 89        Examples
 90        --------
 91
 92        This method is very commonly used for getting "one thing" from CUBE.
 93
 94        ```python
 95        await chris.search_plugins(name_exact="pl-dircopy", version="2.1.1").get_only()
 96        ```
 97
 98        In the example above, a search for plugins given (`name_exact`, `version`)
 99        is guaranteed to return either 0 or 1 result.
100
101        Raises
102        ------
103        chris.util.search.NoneSearchError
104            If this search is empty.
105        chris.util.search.ManySearchError
106            If this search has more than one item and `allow_multiple` is `False`
107
108        See also
109        --------
110        `first` : does the same thing but without checks.
111
112        Parameters
113        ----------
114        allow_multiple: bool
115            if `True`, do not raise `ManySearchError` if `count > 1`
116        """
117        one = await self._get_one()
118        if one.count == 0:
119            raise NoneSearchError(self.url)
120        if not allow_multiple and one.count > 1:
121            raise ManySearchError(self.url)
122        if len(one.results) < 1:
123            raise NonsenseResponseError(
124                f"Response has count={one.count} but the results are empty.", one
125            )
126        return deserialize_linked(self.client, self.Item, one.results[0])
127
128    async def count(self) -> int:
129        """
130        Get the number of items in this collection search.
131
132        Examples
133        --------
134
135        `count` is useful for rendering a progress bar. TODO example with files
136        """
137        one = await self._get_one()
138        return one.count
139
140    async def _get_one(self) -> _Paginated:
141        async with self.client.s.get(self._first_url) as res:
142            await raise_for_status(res)
143            return from_json(_Paginated, await res.text())
144
145    def _paginate(self, url: yarl.URL) -> AsyncIterator[T]:
146        return _get_paginated(
147            client=self.client,
148            url=url,
149            item_type=self.Item,
150            max_requests=self.max_requests,
151        )
152
153    @property
154    def url(self) -> yarl.URL:
155        return self._search_url_with(self.params)
156
157    def _first_aiter(self) -> AsyncIterator[T]:
158        return self._paginate(self._first_url)
159
160    @property
161    def _first_url(self) -> yarl.URL:
162        params = copy.copy(self.params)
163        params["limit"] = 1
164        params["offset"] = 0
165        return self._search_url_with(params)
166
167    @property
168    def _search_url(self) -> yarl.URL:
169        return yarl.URL(self.base_url) / "search/"
170
171    def _search_url_with(self, query: dict[str, Any]):
172        return yarl.URL(self._search_url).with_query(query)

202async def acollect(async_iterable: AsyncIterable[T]) -> list[T]:
203    """
204    Simple helper to convert a `Search` to a [`list`](https://docs.python.org/3/library/stdtypes.html#list).
205
206    Using this function is not recommended unless you can assume the collection is small.
207    """
208    # nb: using tuple here causes
209    #     TypeError: 'async_generator' object is not iterable
210    # return tuple(e async for e in async_iterable)
211    return [e async for e in async_iterable]

 5class Status(enum.Enum):
 6    """
 7    Possible statuses of a plugin instance.
 8    """
 9
10    created = "created"
11    waiting = "waiting"
12    scheduled = "scheduled"
13    started = "started"
14    registeringFiles = "registeringFiles"
15    finishedSuccessfully = "finishedSuccessfully"
16    finishedWithError = "finishedWithError"
17    cancelled = "cancelled"

20class ParameterTypeName(enum.Enum):
21    """
22    Plugin parameter types.
23    """
24
25    string = "string"
26    integer = "integer"
27    float = "float"
28    boolean = "boolean"