Skip to content

Skaha Client

The skaha.client module provides a client for the Skaha server. The client is based of the httpx library and provides a simple interface to the Skaha server. The client configures the authorization headers for user authentication and container registry access.

Bases: BaseModel

Skaha Client.

Parameters:

Name Type Description Default
server str

Server URL.

 required
version str

Skaha API version.

 required
certificate str

Certificate file.

 required
timeout int

HTTP Timeout.

 required
verify bool

Verify SSL certificate.

 required
registry ContainerRegistry

Credentials for a private registry.

 required
client Client

HTTPx Client.

 required
asynclient AsyncClient

HTTPx Async Client.

 required
concurrency int

Number of concurrent requests.

 required
token str

Authentication token.

 required
loglevel int

Logging level. Default is 20 (INFO).

 required

Raises:

Type Description
ValidationError

If the client is misconfigured.
FileNotFoundError

If the cert file does not exist or is not readable.

Examples:

>>> from skaha.client import SkahaClient

class Sample(SkahaClient):
    pass
Source code in skaha/client.py 
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
class SkahaClient(BaseModel):
    """Skaha Client.

    Args:
        server (str): Server URL.
        version (str): Skaha API version.
        certificate (str): Certificate file.
        timeout (int): HTTP Timeout.
        verify (bool): Verify SSL certificate.
        registry (ContainerRegistry): Credentials for a private registry.
        client (Client): HTTPx Client.
        asynclient (AsyncClient): HTTPx Async Client.
        concurrency (int): Number of concurrent requests.
        token (str): Authentication token.
        loglevel (int): Logging level. Default is 20 (INFO).

    Raises:
        ValidationError: If the client is misconfigured.
        FileNotFoundError: If the cert file does not exist or is not readable.

    Examples:
        >>> from skaha.client import SkahaClient

            class Sample(SkahaClient):
                pass
    """

    server: str = Field(
        default="https://ws-uv.canfar.net/skaha",
        title="Skaha Server URL",
        description="Skaha API Server.",
    )
    version: str = Field(
        default="v0",
        title="Skaha API Version",
        description="Version of the Skaha API to use.",
    )
    certificate: str = Field(
        default=f"{environ['HOME']}/.ssl/cadcproxy.pem",
        title="X509 Certificate",
        description="Path to the X509 certificate used for authentication.",
        validate_default=False,
    )
    timeout: int = Field(
        default=15,
        title="HTTP Timeout",
        description="HTTP Timeout.",
    )
    verify: bool = Field(
        default=True,
        title="Verify SSL Certificate[DEPRECATED]",
        description="Whether verify SSL Certs[DEPRECATED].",
        deprecated=True,
    )
    registry: Annotated[
        Optional[ContainerRegistry],
        Field(
            default=None,
            title="Container Registry",
            description="Credentials for a private container registry.",
        ),
    ] = None

    client: Annotated[
        Client,
        Field(
            default=None,
            title="HTTPx Client",
            description="Synchronous HTTPx Client",
            validate_default=False,
            exclude=True,
        ),
    ]

    asynclient: Annotated[
        AsyncClient,
        Field(
            default=None,
            title="HTTPx Client",
            description="Asynchronous HTTPx Client",
            validate_default=False,
            exclude=True,
        ),
    ]

    token: Optional[str] = Field(
        None,
        title="Authentication Token",
        description="Authentication Token for the Skaha Server.",
        exclude=True,
        frozen=True,
    )

    concurrency: int = Field(
        16,
        title="Concurrency",
        description="Number of concurrent requests for the async client.",
        le=256,
        ge=1,
    )

    loglevel: int = Field(
        20,
        title="Logging Level",
        description="10=DEBUG, 20=INFO, 30=WARNING, 40=ERROR, 50=CRITICAL",
        le=50,
        ge=10,
    )

    # Pydantic Configuration
    model_config = ConfigDict(arbitrary_types_allowed=True, extra="forbid")

    @field_validator("loglevel", mode="before")
    @classmethod
    def _set_log_level(cls, value: int) -> int:
        """Set the log level for the client.

        Args:
            value (int): Logging level.

        Returns:
            int: Logging level.
        """
        log.setLevel(value)
        return value

    @field_validator("certificate")
    @classmethod
    def _check_certificate(cls, value: str, info: ValidationInfo) -> str:
        """Validate the certificate file.

        Args:
            value (FilePath): Path to the certificate file.

        Returns:
            FilePath: Validated Path to the certificate file.
        """
        if not info.data.get("token"):
            # Check if the certificate is a valid path
            assert (
                Path(value).resolve(strict=True).is_file()
            ), f"{value} is not a file or does not exist."
            assert access(Path(value), R_OK), f"{value} is not readable."
        return value

    @field_validator("server")
    @classmethod
    def _check_server(cls, value: str) -> str:
        """Validate the server URL.

        Args:
            value (str): Server URL.

        Returns:
            str: Validated Server URL.
        """
        return str(AnyHttpUrl(value))

    @model_validator(mode="after")
    def _configure_clients(self) -> Self:
        """Configure the HTTPx Clients.

        Returns:
            Self: Updated SkahaClient object.
        """
        if self.token:
            self.client, self.asynclient = self._get_token_clients()
        else:
            self.client, self.asynclient = self._get_cert_clients()

        # Configure the HTTP headers
        headers = self._get_headers()
        self.client.headers.update(headers)
        self.asynclient.headers.update(headers)
        # Configure the base URL for the clients
        self.client.base_url = f"{self.server}/{self.version}"
        self.asynclient.base_url = f"{self.server}/{self.version}"
        return self

    def _get_token_clients(self) -> Tuple[Client, AsyncClient]:
        """Get the clients with token authentication.

        Returns:
            Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
        """
        log.info("Using token authentication.")
        client: Client = Client(
            timeout=self.timeout,
        )
        asynclient: AsyncClient = AsyncClient(
            timeout=self.timeout,
        )
        return client, asynclient

    def _get_cert_clients(self) -> Tuple[Client, AsyncClient]:
        """Get the clients with certificate authentication.

        Returns:
            Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
        """
        log.info("Using certificate authentication.")
        ctx = ssl.create_default_context()
        ctx.load_cert_chain(certfile=self.certificate)
        client: Client = Client(
            timeout=self.timeout,
            verify=ctx,
        )
        asynclient: AsyncClient = AsyncClient(
            timeout=self.timeout,
            verify=ctx,
        )
        return client, asynclient

    def _get_headers(self) -> Dict[str, str]:
        """Get the HTTP headers for the client.

        Returns:
            Dict[str, str]: HTTP headers.
        """
        headers: Dict[str, str] = {}
        headers.update({"X-Skaha-Server": str(self.server)})
        headers.update({"Content-Type": "application/x-www-form-urlencoded"})
        headers.update({"Accept": "*/*"})
        headers.update({"Date": asctime(gmtime())})
        headers.update({"X-Skaha-Client": f"python/{__version__}"})
        headers.update({"X-Skaha-Authentication-Type": "certificate"})
        if self.registry:
            headers.update({"X-Skaha-Registry-Auth": f"{self.registry.encoded()}"})
        if self.token:
            headers.update({"Authorization": f"Bearer {self.token}"})
            headers.update({"X-Skaha-Authentication-Type": "token"})
        return headers

_check_certificate(value, info) classmethod

Validate the certificate file.

Parameters:

Name Type Description Default
value FilePath

Path to the certificate file.

 required

Returns:

Name Type Description
FilePath str

Validated Path to the certificate file.
Source code in skaha/client.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
@field_validator("certificate")
@classmethod
def _check_certificate(cls, value: str, info: ValidationInfo) -> str:
    """Validate the certificate file.

    Args:
        value (FilePath): Path to the certificate file.

    Returns:
        FilePath: Validated Path to the certificate file.
    """
    if not info.data.get("token"):
        # Check if the certificate is a valid path
        assert (
            Path(value).resolve(strict=True).is_file()
        ), f"{value} is not a file or does not exist."
        assert access(Path(value), R_OK), f"{value} is not readable."
    return value

_check_server(value) classmethod

Validate the server URL.

Parameters:

Name Type Description Default
value str

Server URL.

 required

Returns:

Name Type Description
str str

Validated Server URL.
Source code in skaha/client.py 
176
177
178
179
180
181
182
183
184
185
186
187
@field_validator("server")
@classmethod
def _check_server(cls, value: str) -> str:
    """Validate the server URL.

    Args:
        value (str): Server URL.

    Returns:
        str: Validated Server URL.
    """
    return str(AnyHttpUrl(value))

_configure_clients()

Configure the HTTPx Clients.

Returns:

Name Type Description
Self Self

Updated SkahaClient object.
Source code in skaha/client.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@model_validator(mode="after")
def _configure_clients(self) -> Self:
    """Configure the HTTPx Clients.

    Returns:
        Self: Updated SkahaClient object.
    """
    if self.token:
        self.client, self.asynclient = self._get_token_clients()
    else:
        self.client, self.asynclient = self._get_cert_clients()

    # Configure the HTTP headers
    headers = self._get_headers()
    self.client.headers.update(headers)
    self.asynclient.headers.update(headers)
    # Configure the base URL for the clients
    self.client.base_url = f"{self.server}/{self.version}"
    self.asynclient.base_url = f"{self.server}/{self.version}"
    return self

_get_cert_clients()

Get the clients with certificate authentication.

Returns:

Type Description
Tuple[Client, AsyncClient]

Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
Source code in skaha/client.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def _get_cert_clients(self) -> Tuple[Client, AsyncClient]:
    """Get the clients with certificate authentication.

    Returns:
        Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
    """
    log.info("Using certificate authentication.")
    ctx = ssl.create_default_context()
    ctx.load_cert_chain(certfile=self.certificate)
    client: Client = Client(
        timeout=self.timeout,
        verify=ctx,
    )
    asynclient: AsyncClient = AsyncClient(
        timeout=self.timeout,
        verify=ctx,
    )
    return client, asynclient

_get_headers()

Get the HTTP headers for the client.

Returns:

Type Description
Dict[str, str]

Dict[str, str]: HTTP headers.
Source code in skaha/client.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def _get_headers(self) -> Dict[str, str]:
    """Get the HTTP headers for the client.

    Returns:
        Dict[str, str]: HTTP headers.
    """
    headers: Dict[str, str] = {}
    headers.update({"X-Skaha-Server": str(self.server)})
    headers.update({"Content-Type": "application/x-www-form-urlencoded"})
    headers.update({"Accept": "*/*"})
    headers.update({"Date": asctime(gmtime())})
    headers.update({"X-Skaha-Client": f"python/{__version__}"})
    headers.update({"X-Skaha-Authentication-Type": "certificate"})
    if self.registry:
        headers.update({"X-Skaha-Registry-Auth": f"{self.registry.encoded()}"})
    if self.token:
        headers.update({"Authorization": f"Bearer {self.token}"})
        headers.update({"X-Skaha-Authentication-Type": "token"})
    return headers

_get_token_clients()

Get the clients with token authentication.

Returns:

Type Description
Tuple[Client, AsyncClient]

Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
Source code in skaha/client.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
def _get_token_clients(self) -> Tuple[Client, AsyncClient]:
    """Get the clients with token authentication.

    Returns:
        Tuple[Client, AsyncClient]: Synchronous and Asynchronous HTTPx Clients.
    """
    log.info("Using token authentication.")
    client: Client = Client(
        timeout=self.timeout,
    )
    asynclient: AsyncClient = AsyncClient(
        timeout=self.timeout,
    )
    return client, asynclient

_set_log_level(value) classmethod

Set the log level for the client.

Parameters:

Name Type Description Default
value int

Logging level.

 required

Returns:

Name Type Description
int int

Logging level.
Source code in skaha/client.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
@field_validator("loglevel", mode="before")
@classmethod
def _set_log_level(cls, value: int) -> int:
    """Set the log level for the client.

    Args:
        value (int): Logging level.

    Returns:
        int: Logging level.
    """
    log.setLevel(value)
    return value