jumpstarter-python/packages/jumpstarter-driver-opendal/jumpstarter_driver_opendal/common.py at e234f159a659a30f8ebbdef6d4cb867699b05b20 · jumpstarter-dev/jumpstarter-python · 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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
# Reference: https://github.com/apache/opendal/blob/main/bindings/python/python/opendal/__init__.pyi
import warnings
from os import PathLike
from typing import Any, Literal, Optional

import opendal
from pydantic import BaseModel, Field, model_validator

warnings.filterwarnings(
    "ignore", 'Field name "copy" in "Capability" shadows an attribute in parent "BaseModel"', UserWarning
)

Mode = Literal["rb", "wb"]
HashAlgo = Literal["md5", "sha256"]
PathBuf = str | PathLike


class EntryMode(BaseModel):
    entry_is_file: bool = Field(serialization_alias="is_file")
    entry_is_dir: bool = Field(serialization_alias="is_dir")

    @model_validator(mode="before")
    @classmethod
    def __validate(cls, data: Any):
        match data:
            case opendal.EntryMode():
                return EntryMode(
                    entry_is_file=data.is_file(),
                    entry_is_dir=data.is_dir(),
                )
            case _:
                return data

    def is_file(self) -> bool:
        return self.entry_is_file

    def is_dir(self) -> bool:
        return self.entry_is_dir


class Metadata(BaseModel):
    content_disposition: Optional[str]
    content_length: int
    content_md5: Optional[str]
    content_type: Optional[str]
    etag: Optional[str]
    mode: EntryMode


class PresignedRequest(BaseModel):
    """
    Presigned HTTP request

    Allows you to delegate access to a specific file in your storage backend
    without sharing access credentials
    """

    url: str
    """
    HTTP request URL
    """
    method: str
    """
    HTTP method

        GET: download file

        PUT: upload file

        DELETE: delete file
    """
    headers: dict[str, str]
    """
    Additional HTTP headers to send with the request
    """


class Capability(BaseModel):
    stat: bool
    stat_with_if_match: bool
    stat_with_if_none_match: bool

    read: bool
    read_with_if_match: bool
    read_with_if_none_match: bool
    read_with_if_modified_since: bool
    read_with_if_unmodified_since: bool
    read_with_override_cache_control: bool
    read_with_override_content_disposition: bool
    read_with_override_content_type: bool
    read_with_version: bool

    write: bool
    write_can_multi: bool
    write_can_empty: bool
    write_can_append: bool
    write_with_content_type: bool
    write_with_content_disposition: bool
    write_with_content_encoding: bool
    write_with_cache_control: bool
    write_with_if_match: bool
    write_with_if_none_match: bool
    write_with_if_not_exists: bool
    write_with_user_metadata: bool
    write_multi_max_size: int | None
    write_multi_min_size: int | None
    write_total_max_size: int | None

    create_dir: bool
    delete: bool
    copy: bool
    rename: bool

    list: bool
    list_with_limit: bool
    list_with_start_after: bool
    list_with_recursive: bool

    presign: bool
    presign_read: bool
    presign_stat: bool
    presign_write: bool
    presign_delete: bool

    shared: bool