aiochris.models.public

Read-only models for CUBE resources.

 1"""
 2Read-only models for CUBE resources.
 3"""
 4
 5import sys
 6from dataclasses import dataclass
 7from typing import Optional, Literal, TextIO
 8
 9import serde
10
11from aiochris.enums import PluginType
12from aiochris.link import http
13from aiochris.link.linked import LinkedModel
14from aiochris.types import *
15from aiochris.util.search import Search
16
17
18@serde.serde
19@dataclass(frozen=True)
20class ComputeResource:
21    url: ApiUrl
22    id: ComputeResourceId
23    creation_date: str
24    modification_date: str
25    name: ComputeResourceName
26    compute_url: PfconUrl
27    compute_auth_url: str
28    description: str
29    max_job_exec_seconds: int
30
31
32@serde.serde
33@dataclass(frozen=True)
34class PluginParameter(LinkedModel):
35    """
36    Information about a parameter (a command-line option/flag) of a plugin.
37    """
38
39    url: PluginParameterUrl
40    id: PluginParameterId
41    name: ParameterName
42    type: ParameterType
43    optional: bool
44    default: Optional[ParameterValue]
45    flag: str
46    short_flag: str
47    action: Literal["store", "store_true", "store_false"]
48    help: str
49    ui_exposed: bool
50    plugin: PluginUrl
51
52
53@serde.serde
54@dataclass(frozen=True)
55class PublicPlugin(LinkedModel):
56    """
57    A ChRIS plugin.
58    """
59
60    url: PluginUrl
61    id: PluginId
62    name: PluginName
63    version: PluginVersion
64    dock_image: ImageTag
65    public_repo: str
66    compute_resources: ComputeResourceUrl
67    parameters: PluginParametersUrl
68    plugin_type: PluginType = serde.field(rename="type")
69
70    @http.search("compute_resources", subpath="")
71    def get_compute_resources(self) -> Search[ComputeResource]:
72        """Get the compute resources this plugin is registered to."""
73        ...
74
75    @http.search("parameters", subpath="")
76    def get_parameters(self) -> Search[PluginParameter]:
77        """Get the parameters of this plugin."""
78        ...
79
80    async def print_help(self, out: TextIO = sys.stdout) -> None:
81        """
82        Display the help messages for this plugin's parameters.
83        """
84        async for param in self.get_parameters():
85            left = f"{param.name} ({param.flag})"
86            out.write(f"{left:>20}: {param.help}")
87            if param.default is not None:
88                out.write(f" (default: {param.default})")
89            out.write("\n")
@serde.serde
@dataclass(frozen=True)
class ComputeResource: 
19@serde.serde
20@dataclass(frozen=True)
21class ComputeResource:
22    url: ApiUrl
23    id: ComputeResourceId
24    creation_date: str
25    modification_date: str
26    name: ComputeResourceName
27    compute_url: PfconUrl
28    compute_auth_url: str
29    description: str
30    max_job_exec_seconds: int
ComputeResource( url: aiochris.types.ApiUrl, id: aiochris.types.ComputeResourceId, creation_date: str, modification_date: str, name: aiochris.types.ComputeResourceName, compute_url: aiochris.types.PfconUrl, compute_auth_url: str, description: str, max_job_exec_seconds: int)
creation_date: str
modification_date: str
compute_auth_url: str
description: str
max_job_exec_seconds: int
@serde.serde
@dataclass(frozen=True)
class PluginParameter(aiochris.link.linked.LinkedModel): 
33@serde.serde
34@dataclass(frozen=True)
35class PluginParameter(LinkedModel):
36    """
37    Information about a parameter (a command-line option/flag) of a plugin.
38    """
39
40    url: PluginParameterUrl
41    id: PluginParameterId
42    name: ParameterName
43    type: ParameterType
44    optional: bool
45    default: Optional[ParameterValue]
46    flag: str
47    short_flag: str
48    action: Literal["store", "store_true", "store_false"]
49    help: str
50    ui_exposed: bool
51    plugin: PluginUrl

Information about a parameter (a command-line option/flag) of a plugin.

PluginParameter( s: aiohttp.client.ClientSession, max_search_requests: int, url: aiochris.types.PluginParameterUrl, id: aiochris.types.ParameterGlobalId, name: aiochris.types.ParameterName, type: Literal['boolean', 'integer', 'float', 'string', 'path', 'unextpath'], optional: bool, default: Union[str, int, float, bool, NoneType], flag: str, short_flag: str, action: Literal['store', 'store_true', 'store_false'], help: str, ui_exposed: bool, plugin: aiochris.types.PluginUrl)
id: aiochris.types.ParameterGlobalId
type: Literal['boolean', 'integer', 'float', 'string', 'path', 'unextpath']
optional: bool
default: Union[str, int, float, bool, NoneType]
flag: str
short_flag: str
action: Literal['store', 'store_true', 'store_false']
help: str
ui_exposed: bool
Inherited Members
aiochris.link.linked.LinkedModel
to_dict
aiochris.link.linked.Linked
s
max_search_requests
@serde.serde
@dataclass(frozen=True)
class PublicPlugin(aiochris.link.linked.LinkedModel): 
54@serde.serde
55@dataclass(frozen=True)
56class PublicPlugin(LinkedModel):
57    """
58    A ChRIS plugin.
59    """
60
61    url: PluginUrl
62    id: PluginId
63    name: PluginName
64    version: PluginVersion
65    dock_image: ImageTag
66    public_repo: str
67    compute_resources: ComputeResourceUrl
68    parameters: PluginParametersUrl
69    plugin_type: PluginType = serde.field(rename="type")
70
71    @http.search("compute_resources", subpath="")
72    def get_compute_resources(self) -> Search[ComputeResource]:
73        """Get the compute resources this plugin is registered to."""
74        ...
75
76    @http.search("parameters", subpath="")
77    def get_parameters(self) -> Search[PluginParameter]:
78        """Get the parameters of this plugin."""
79        ...
80
81    async def print_help(self, out: TextIO = sys.stdout) -> None:
82        """
83        Display the help messages for this plugin's parameters.
84        """
85        async for param in self.get_parameters():
86            left = f"{param.name} ({param.flag})"
87            out.write(f"{left:>20}: {param.help}")
88            if param.default is not None:
89                out.write(f" (default: {param.default})")
90            out.write("\n")

A ChRIS plugin.

PublicPlugin( s: aiohttp.client.ClientSession, max_search_requests: int, url: aiochris.types.PluginUrl, id: aiochris.types.PluginId, name: aiochris.types.PluginName, version: aiochris.types.PluginVersion, dock_image: aiochris.types.ImageTag, public_repo: str, compute_resources: aiochris.types.ComputeResourceUrl, parameters: aiochris.types.PluginParametersUrl, plugin_type: aiochris.enums.PluginType)
public_repo: str
plugin_type: aiochris.enums.PluginType
@http.search('compute_resources', subpath='')
def get_compute_resources( self) -> aiochris.util.search.Search[ComputeResource]: 
71    @http.search("compute_resources", subpath="")
72    def get_compute_resources(self) -> Search[ComputeResource]:
73        """Get the compute resources this plugin is registered to."""
74        ...

Get the compute resources this plugin is registered to.

@http.search('parameters', subpath='')
def get_parameters( self) -> aiochris.util.search.Search[PluginParameter]: 
76    @http.search("parameters", subpath="")
77    def get_parameters(self) -> Search[PluginParameter]:
78        """Get the parameters of this plugin."""
79        ...

Get the parameters of this plugin.

async def print_help(self, out: <class 'TextIO'> = <_io.StringIO object>) -> None: 
81    async def print_help(self, out: TextIO = sys.stdout) -> None:
82        """
83        Display the help messages for this plugin's parameters.
84        """
85        async for param in self.get_parameters():
86            left = f"{param.name} ({param.flag})"
87            out.write(f"{left:>20}: {param.help}")
88            if param.default is not None:
89                out.write(f" (default: {param.default})")
90            out.write("\n")

Display the help messages for this plugin's parameters.

Inherited Members
aiochris.link.linked.LinkedModel
to_dict
aiochris.link.linked.Linked
s
max_search_requests