backend.ai/src/ai/backend/manager/services/image/actions/purge_images.py at 067875675a750b1be87e60b71e0857465e72dab3 · lablup/backend.ai · GitHub

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
from dataclasses import dataclass
from typing import override

from ai.backend.common.data.permission.types import RBACElementType
from ai.backend.common.types import AgentId, ImageID
from ai.backend.manager.actions.action import BaseActionResult
from ai.backend.manager.actions.types import ActionOperationType
from ai.backend.manager.data.image.types import ImageData
from ai.backend.manager.data.permission.types import RBACElementRef
from ai.backend.manager.services.image.actions.base import (
    ImageAction,
    ImageSingleEntityAction,
    ImageSingleEntityActionResult,
)
from ai.backend.manager.services.image.types import ImageRefData


@dataclass
class PurgeImageAction(ImageAction):
    image: ImageRefData
    agent_id: AgentId
    force: bool
    noprune: bool

    @override
    def entity_id(self) -> str | None:
        return None

    @override
    @classmethod
    def operation_type(cls) -> ActionOperationType:
        return ActionOperationType.PURGE


@dataclass
class PurgeImageActionResult(BaseActionResult):
    reserved_bytes: int
    purged_image: ImageData
    error: str | None

    @override
    def entity_id(self) -> str | None:
        return str(self.purged_image.id)


@dataclass
class PurgedImagesData:
    agent_id: AgentId
    purged_images: list[str]


@dataclass
class PurgeImagesKeyData:
    agent_id: AgentId
    images: list[ImageRefData]


# TODO: Remove this?
@dataclass
class PurgeImagesAction(ImageAction):
    keys: list[PurgeImagesKeyData]
    force: bool
    noprune: bool

    @override
    def entity_id(self) -> str | None:
        return None

    @override
    @classmethod
    def operation_type(cls) -> ActionOperationType:
        return ActionOperationType.PURGE


@dataclass
class PurgeImagesActionResult(BaseActionResult):
    total_reserved_bytes: int
    purged_images: list[PurgedImagesData]
    errors: list[str]

    @override
    def entity_id(self) -> str | None:
        return None


@dataclass
class PurgeImageByIdAction(ImageSingleEntityAction):
    image_id: ImageID

    @override
    @classmethod
    def operation_type(cls) -> ActionOperationType:
        return ActionOperationType.PURGE

    @override
    def target_entity_id(self) -> str:
        return str(self.image_id)

    @override
    def target_element(self) -> RBACElementRef:
        return RBACElementRef(RBACElementType.IMAGE, str(self.image_id))


@dataclass
class PurgeImageByIdActionResult(ImageSingleEntityActionResult):
    image: ImageData

    @override
    def target_entity_id(self) -> str:
        return str(self.image.id)