-
Notifications
You must be signed in to change notification settings - Fork 2
/
api.py
443 lines (313 loc) · 12.8 KB
/
api.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
"""
Functions for interacting with the snapd REST API.
See https://snapcraft.io/docs/snapd-api for documentation of the API.
Permissions are based on the user calling the API, most mutative interactions
(install, refresh, etc) require root.
"""
from typing import Any, Dict, List, Literal, Optional, Union
from . import http
from .types import AssertionData, FileUpload, FormData, SnapdResponse
def check_change(cid: str) -> SnapdResponse:
"""Checks the status of snapd change with id `cid`."""
return http.get("/changes/" + cid)
def check_changes() -> SnapdResponse:
"""Checks the status of all snapd changes."""
return http.get("/changes?select=all")
def enable(name: str) -> SnapdResponse:
"""Enables a previously disabled snap by `name`."""
return http.post("/snaps/" + name, {"action": "enable"})
def enable_all(names: List[str]) -> SnapdResponse:
"""Like `enable_snap`, but for the list of snaps in `names`.
NOTE: as of 2024-01-08, enable/disable is not yet supported for multiple snaps.
"""
return http.post("/snaps", {"action": "enable", "snaps": names})
def disable(name: str) -> SnapdResponse:
"""Disables a snap by `name`, making its binaries and services unavailable."""
return http.post("/snaps/" + name, {"action": "disable"})
def disable_all(names: List[str]) -> SnapdResponse:
"""Like `disable_snap`, but for the list of snaps in `names`.
NOTE: as of 2024-01-08, enable/disable is not yet supported for multiple snaps.
"""
return http.post("/snaps", {"action": "disable", "snaps": names})
def hold(
name: str,
*,
hold_level: Literal["general", "auto-refresh"] = "general",
time: str = "forever",
) -> SnapdResponse:
"""Holds a snap by `name` at `hold_level` until `time`.
:param time: RFC3339 timestamp to hold the snap until, or "forever".
"""
return http.post(
"/snaps/" + name, {"action": "hold", "hold-level": hold_level, "time": time}
)
def hold_all(
names: List[str],
*,
hold_level: Literal["general", "auto-refresh"] = "general",
time: str = "forever",
) -> SnapdResponse:
"""Like `hold_snap`, but for the list of snaps in `names`."""
return http.post(
"/snaps",
{"action": "hold", "snaps": names, "hold-level": hold_level, "time": time},
)
def install(
name: str,
*,
revision: Optional[str] = None,
channel: Optional[str] = None,
classic: bool = False,
) -> SnapdResponse:
"""Installs a snap by `name` at `revision`, tracking `channel`.
:param revision: revision to install. Defaults to latest.
:param channel: channel to track. Defaults to stable.
:param classic: if `True`, snap is installed in classic containment mode.
"""
body: Dict[str, Union[str, bool]] = {"action": "install"}
if revision is not None:
body["revision"] = revision
if channel is not None:
body["channel"] = channel
if classic:
body["classic"] = classic
return http.post("/snaps/" + name, body)
def install_all(names: List[str]) -> SnapdResponse:
"""Installs all snaps in `names` using the latest rev of the stable channel, with strict
confinement.
"""
return http.post("/snaps", {"action": "install", "snaps": names})
def sideload(
file_paths: List[str],
*,
classic: bool = False,
dangerous: bool = False,
devmode: bool = False,
jailmode: bool = False,
system_restart_immediate: bool = False,
) -> SnapdResponse:
"""Sideload a snap from the local filesystem.
:param file_paths: Paths to the snap files to install.
:param classic: if true, put snaps in classic mode and disable
security confinement
:param dangerous: if true, install the given snap files even if there are
no pre-acknowledged signatures for them
:param devmode: if true, put snaps in development mode and disable
security confinement
:param jailmode: if true, put snaps in enforced confinement mode
:param system_restart_immediate: if true, makes any system restart,
immediately and without delay (requires snapd 2.52)
"""
data: Dict[str, Union[str, bool]] = {"action": "install"}
if classic:
data["classic"] = classic
if dangerous:
data["dangerous"] = dangerous
if devmode:
data["devmode"] = devmode
if jailmode:
data["jailmode"] = jailmode
if system_restart_immediate:
data["system-restart-immediate"] = system_restart_immediate
files = [FileUpload(name="snap", path=file_path) for file_path in file_paths]
return http.post("/snaps", FormData(data=data, files=files))
def refresh(
name: str,
*,
revision: Optional[str] = None,
channel: Optional[str] = None,
classic: bool = False,
) -> SnapdResponse:
"""Refreshes a snap by `name`, to `revision`, tracking `channel`.
:param revision: revision to refresh to. Defaults to latest.
:param channel: channel to switch tracking to. Default to stable.
:param classic: If `True`, snap is changed to classic containment mode.
"""
body: Dict[str, Union[str, bool]] = {"action": "refresh"}
if revision is not None:
body["revision"] = revision
if channel is not None:
body["channel"] = channel
if classic:
body["classic"] = classic
return http.post("/snaps/" + name, body)
def refresh_all(names: Optional[List[str]] = None) -> SnapdResponse:
"""Refreshes all snaps in `names` to the latest revision. If `names` is not provided or empty,
all snaps are refreshed.
"""
body: Dict[str, Union[str, List[str]]] = {"action": "refresh"}
if names:
body["snaps"] = names
return http.post("/snaps", body)
def revert(
name: str, *, revision: Optional[str] = None, classic: Optional[bool] = None
) -> SnapdResponse:
"""Reverts a snap, switching what revision is currently installed.
:param revision: If provided, the revision to switch to. Otherwise, the revision used prior to
the last refresh is used.
:param classic: If `True`, confinement is changed to classic. If `False`, confinement is
changed to strict. If not provided, confinement is left as-is.
"""
body: Dict[str, Union[str, bool]] = {"action": "revert"}
if revision is not None:
body["revision"] = revision
if classic is not None:
body["classic"] = classic
return http.post("/snaps/" + name, body)
def revert_all(names: List[str]) -> SnapdResponse:
"""Reverts all snaps in `names` to the revision used prior to the last refresh."""
return http.post("/snaps", {"action": "revert", "snaps": names})
def remove(name: str) -> SnapdResponse:
"""Uninstalls a snap identified by `name`."""
return http.post("/snaps/" + name, {"action": "remove"})
def remove_all(names: List[str]) -> SnapdResponse:
"""Uninstalls all snaps identified in `names`."""
return http.post("/snaps", {"action": "remove", "snaps": names})
def switch(name: str, *, channel: str = "stable") -> SnapdResponse:
"""Switches the tracking channel of snap `name`."""
return http.post("/snaps/" + name, {"action": "switch", "channel": channel})
def switch_all(names: List[str], channel: str = "stable") -> SnapdResponse:
"""Switches the tracking channels of all snaps in `names`.
NOTE: as of 2024-01-08, switch is not yet supported for multiple snaps.
"""
return http.post("/snaps", {"action": "switch", "channel": channel, "snaps": names})
def unhold(name: str) -> SnapdResponse:
"""Removes the hold on a snap, allowing it to refresh on its usual schedule."""
return http.post("/snaps/" + name, {"action": "unhold"})
def unhold_all(names: List[str]) -> SnapdResponse:
"""Removes the holds on all snaps in `names`, allowing them to refresh on their usual
schedule.
"""
return http.post("/snaps", {"action": "unhold", "snaps": names})
def list() -> SnapdResponse:
"""GETs a list of installed snaps.
This stomps on builtins.list, so please import it namespaced.
"""
return http.get("/snaps")
# Configuration: get and set snap options
def get_conf(name: str, *, keys: Optional[List[str]] = None) -> SnapdResponse:
"""Get the configuration details for the snap `name`.
:param name: the name of the snap.
:param keys: retrieve the configuration for these specific `keys`. Dotted
keys can be used to retrieve nested values.
"""
query_params = {}
if keys:
query_params["keys"] = ",".join(keys)
return http.get(f"/snaps/{name}/conf", query_params=query_params)
def set_conf(name: str, config: Dict[str, Any]) -> SnapdResponse:
"""Set the configuration details for the snap `name`.
:param name: the name of the snap.
:param config: A key-value mapping of snap configuration.
Keys can be dotted, `None` can be used to unset config options.
"""
return http.put(f"/snaps/{name}/conf", config)
# Assertions: list and add assertions
def get_assertion_types() -> SnapdResponse:
"""GETs the list of assertion types."""
return http.get("/assertions")
def get_assertions(
assertion_type: str, filters: Optional[Dict[str, Any]] = None
) -> SnapdResponse:
"""GETs all the assertions of the given type.
The response is a stream of assertions separated by double newlines.
:param assertion_type: The type of the assertion.
:param filters: A (assertion-header, filter-value) mapping to filter
assertions with. Examples of headers are: username, authority-id,
account-id, series, publisher, snap-name, and publisher-id.
"""
return http.get(f"/assertions/{assertion_type}", query_params=filters)
def add_assertion(assertion: str) -> SnapdResponse:
"""Add an assertion to the system assertion database.
:param assertion: The assertion to add. It may also be a newer revision
of a pre-existing assertion that it will replace.
"""
body = AssertionData(assertion)
return http.post("/assertions", body)
# Users
def list_users() -> SnapdResponse:
"""Get information on user accounts."""
return http.get("/users")
def add_user(
username: str,
email: str,
sudoer: bool = False,
known: bool = False,
force_managed: bool = False,
automatic: bool = False,
) -> SnapdResponse:
"""Create a local user."""
body = {
"action": "create",
"username": username,
"email": email,
"sudoer": sudoer,
"known": known,
"force-managed": force_managed,
"automatic": automatic,
}
return http.post("/users", body)
def remove_user(username: str) -> SnapdResponse:
"""Remove a local user."""
body = {"action": "remove", "username": username}
return http.post("/users", body)
# Apps and Services
def get_apps(names: List[str] = [], services_only: bool = False):
"""List available apps.
:param services_only: Return only services.
:param names: List apps for the snaps in `names` only.
"""
query_params = {}
if services_only:
query_params["select"] = "service"
if names:
query_params["names"] = ",".join(names)
return http.get("/apps", query_params=query_params)
def start(name: str, enable: bool = False) -> SnapdResponse:
"""Start the service `name`.
:param enable: arranges to have the service start at system start.
"""
return http.post(
"/apps",
{"action": "start", "names": [name], "enable": enable},
)
def start_all(names: List[str], enable: bool = False) -> SnapdResponse:
"""Start the services in `names`.
:param enable: arranges to have the service start at system start.
"""
return http.post(
"/apps",
{"action": "start", "names": names, "enable": enable},
)
def stop(name: str, disable: bool = False) -> SnapdResponse:
"""Stop the service `name`.
:param disable: arranges to no longer start the service at system start.
"""
return http.post(
"/apps",
{"action": "stop", "names": [name], "disable": disable},
)
def stop_all(names: List[str], disable: bool = False) -> SnapdResponse:
"""Stop the services in `names`.
:param disable: arranges to no longer start the service at system start.
"""
return http.post(
"/apps",
{"action": "stop", "names": names, "disable": disable},
)
def restart(name: str, reload: bool = False) -> SnapdResponse:
"""Restart the service `name`.
:param reload: try to reload the service instead of restarting.
"""
return http.post(
"/apps",
{"action": "restart", "names": [name], "reload": reload},
)
def restart_all(names: List[str], reload: bool = False) -> SnapdResponse:
"""Restart the services in `names`.
:param reload: try to reload the service instead of restarting.
"""
return http.post(
"/apps",
{"action": "restart", "names": names, "reload": reload},
)