Skip to content

Client

HTTP Client for Reduct Storage HTTP API

Source code in reduct/client.py 
 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
class Client:
    """HTTP Client for Reduct Storage HTTP API"""

    def __init__(
        self,
        url: str,
        api_token: Optional[str] = None,
        timeout: Optional[float] = None,
        extra_headers: Optional[Dict[str, str]] = None,
        **kwargs,
    ):
        """
        Constructor

        Args:
            url: URL to connect to the storage
            api_token: API token if the storage uses it for authorization
            timeout: total timeout for connection, request and response in seconds
            extra_headers: extra headers to send with each request
        Kwargs:
            session: an external aiohttp session to use for requests
        Examples:
            >>> client = Client("http://127.0.0.1:8383")
            >>> info = await client.info()
        """
        self._http = HttpClient(
            url.rstrip("/"), api_token, timeout, extra_headers, **kwargs
        )

    async def __aenter__(self):
        self._http._session = ClientSession(timeout=self._http._timeout)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        return await self._http._session.close()

    async def info(self) -> ServerInfo:
        """
        Get high level server info

        Returns:
            ServerInfo:

        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all("GET", "/info")
        return ServerInfo.model_validate_json(body)

    async def list(self) -> List[BucketInfo]:
        """
        Return a list of all buckets on server

        Returns:
            List[BucketInfo]
        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all("GET", "/list")
        return BucketList.model_validate_json(body).buckets

    async def get_bucket(self, name: str) -> Bucket:
        """
        Load a bucket to work with
        Args:
            name: name of the bucket
        Returns:
            Bucket
        Raises:
            ReductError: if there is an HTTP error
        """
        await self._http.request_all("GET", f"/b/{name}")
        return Bucket(name, self._http)

    async def create_bucket(
        self,
        name: str,
        settings: Optional[BucketSettings] = None,
        exist_ok: bool = False,
    ) -> Bucket:
        """
        Create a new bucket
        Args:
            name: a name for the bucket
            settings: settings for the bucket If None, the server
            default settings is used.
            exist_ok: the client raises no exception if the bucket
                already exists and returns it
        Returns:
            Bucket: created bucket
        Raises:
            ReductError: if there is an HTTP error
        """
        data = settings.model_dump_json() if settings else None
        try:
            await self._http.request_all("POST", f"/b/{name}", data=data)
        except ReductError as err:
            if err.status_code != 409 or not exist_ok:
                raise err

        return Bucket(name, self._http)

    async def get_token_list(self) -> List[Token]:
        """
        Get a list of all tokens
        Returns:
            List[Token]
        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all("GET", "/tokens")
        return TokenList.model_validate_json(body).tokens

    async def get_token(self, name: str) -> FullTokenInfo:
        """
        Get a token by name
        Args:
            name: name of the token
        Returns:
            Token
        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all("GET", f"/tokens/{name}")
        return FullTokenInfo.model_validate_json(body)

    async def create_token(self, name: str, permissions: Permissions) -> str:
        """
        Create a new token
        Args:
            name: name of the token
            permissions: permissions for the token
        Returns:
            str: token value
        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all(
            "POST", f"/tokens/{name}", data=permissions.model_dump_json()
        )
        return TokenCreateResponse.model_validate_json(body).value

    async def remove_token(self, name: str) -> None:
        """
        Delete a token
        Args:
            name: name of the token
        Raises:
            ReductError: if there is an HTTP error
        """
        await self._http.request_all("DELETE", f"/tokens/{name}")

    async def me(self) -> FullTokenInfo:
        """
        Get information about the current token
        Returns:
            FullTokenInfo
        Raises:
            ReductError: if there is an HTTP error
        """
        body, _ = await self._http.request_all("GET", "/me")
        return FullTokenInfo.model_validate_json(body)

__init__(url, api_token=None, timeout=None, extra_headers=None, **kwargs)

Constructor

Parameters:

Name Type Description Default
url str

URL to connect to the storage

 required
api_token Optional[str]

API token if the storage uses it for authorization

 None
timeout Optional[float]

total timeout for connection, request and response in seconds

 None
extra_headers Optional[Dict[str, str]]

extra headers to send with each request

 None

Kwargs: session: an external aiohttp session to use for requests Examples: >>> client = Client("http://127.0.0.1:8383") >>> info = await client.info()

Source code in reduct/client.py 
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
def __init__(
    self,
    url: str,
    api_token: Optional[str] = None,
    timeout: Optional[float] = None,
    extra_headers: Optional[Dict[str, str]] = None,
    **kwargs,
):
    """
    Constructor

    Args:
        url: URL to connect to the storage
        api_token: API token if the storage uses it for authorization
        timeout: total timeout for connection, request and response in seconds
        extra_headers: extra headers to send with each request
    Kwargs:
        session: an external aiohttp session to use for requests
    Examples:
        >>> client = Client("http://127.0.0.1:8383")
        >>> info = await client.info()
    """
    self._http = HttpClient(
        url.rstrip("/"), api_token, timeout, extra_headers, **kwargs
    )

create_bucket(name, settings=None, exist_ok=False) async

Create a new bucket Args: name: a name for the bucket settings: settings for the bucket If None, the server default settings is used. exist_ok: the client raises no exception if the bucket already exists and returns it Returns: Bucket: created bucket Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
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
async def create_bucket(
    self,
    name: str,
    settings: Optional[BucketSettings] = None,
    exist_ok: bool = False,
) -> Bucket:
    """
    Create a new bucket
    Args:
        name: a name for the bucket
        settings: settings for the bucket If None, the server
        default settings is used.
        exist_ok: the client raises no exception if the bucket
            already exists and returns it
    Returns:
        Bucket: created bucket
    Raises:
        ReductError: if there is an HTTP error
    """
    data = settings.model_dump_json() if settings else None
    try:
        await self._http.request_all("POST", f"/b/{name}", data=data)
    except ReductError as err:
        if err.status_code != 409 or not exist_ok:
            raise err

    return Bucket(name, self._http)

create_token(name, permissions) async

Create a new token Args: name: name of the token permissions: permissions for the token Returns: str: token value Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
async def create_token(self, name: str, permissions: Permissions) -> str:
    """
    Create a new token
    Args:
        name: name of the token
        permissions: permissions for the token
    Returns:
        str: token value
    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all(
        "POST", f"/tokens/{name}", data=permissions.model_dump_json()
    )
    return TokenCreateResponse.model_validate_json(body).value

get_bucket(name) async

Load a bucket to work with Args: name: name of the bucket Returns: Bucket Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
158
159
160
161
162
163
164
165
166
167
168
169
async def get_bucket(self, name: str) -> Bucket:
    """
    Load a bucket to work with
    Args:
        name: name of the bucket
    Returns:
        Bucket
    Raises:
        ReductError: if there is an HTTP error
    """
    await self._http.request_all("GET", f"/b/{name}")
    return Bucket(name, self._http)

get_token(name) async

Get a token by name Args: name: name of the token Returns: Token Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
210
211
212
213
214
215
216
217
218
219
220
221
async def get_token(self, name: str) -> FullTokenInfo:
    """
    Get a token by name
    Args:
        name: name of the token
    Returns:
        Token
    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all("GET", f"/tokens/{name}")
    return FullTokenInfo.model_validate_json(body)

get_token_list() async

Get a list of all tokens Returns: List[Token] Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
199
200
201
202
203
204
205
206
207
208
async def get_token_list(self) -> List[Token]:
    """
    Get a list of all tokens
    Returns:
        List[Token]
    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all("GET", "/tokens")
    return TokenList.model_validate_json(body).tokens

info() async

Get high level server info

Returns:

Name Type Description
ServerInfo ServerInfo

Raises:

Type Description
ReductError

if there is an HTTP error
Source code in reduct/client.py 
133
134
135
136
137
138
139
140
141
142
143
144
async def info(self) -> ServerInfo:
    """
    Get high level server info

    Returns:
        ServerInfo:

    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all("GET", "/info")
    return ServerInfo.model_validate_json(body)

list() async

Return a list of all buckets on server

Returns:

Type Description
List[BucketInfo]

List[BucketInfo]

Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
146
147
148
149
150
151
152
153
154
155
156
async def list(self) -> List[BucketInfo]:
    """
    Return a list of all buckets on server

    Returns:
        List[BucketInfo]
    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all("GET", "/list")
    return BucketList.model_validate_json(body).buckets

me() async

Get information about the current token Returns: FullTokenInfo Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
249
250
251
252
253
254
255
256
257
258
async def me(self) -> FullTokenInfo:
    """
    Get information about the current token
    Returns:
        FullTokenInfo
    Raises:
        ReductError: if there is an HTTP error
    """
    body, _ = await self._http.request_all("GET", "/me")
    return FullTokenInfo.model_validate_json(body)

remove_token(name) async

Delete a token Args: name: name of the token Raises: ReductError: if there is an HTTP error

Source code in reduct/client.py 
239
240
241
242
243
244
245
246
247
async def remove_token(self, name: str) -> None:
    """
    Delete a token
    Args:
        name: name of the token
    Raises:
        ReductError: if there is an HTTP error
    """
    await self._http.request_all("DELETE", f"/tokens/{name}")