Repository navigation
Expand file tree
/
Copy pathuploads.py
More file actions
67 lines (51 loc) · 1.8 KB
/
Copy pathuploads.py
File metadata and controls
67 lines (51 loc) · 1.8 KB
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
"""Two-step upload flow: presign + PUT."""
from __future__ import annotations
from typing import TYPE_CHECKING
from stemsplit_python.models.uploads import UploadTicket
if TYPE_CHECKING:
from stemsplit_python._transport import Transport
class UploadsResource:
"""``POST /upload`` and the streaming PUT to the resulting presigned URL."""
def __init__(self, transport: Transport) -> None:
self._transport = transport
def create_ticket(
self,
*,
filename: str,
content_type: str | None = None,
) -> UploadTicket:
"""Request a presigned upload URL.
The returned ticket is valid for 15 minutes. Use :attr:`UploadTicket.upload_key`
when calling ``jobs.create``.
"""
body = {"filename": filename}
if content_type:
body["contentType"] = content_type
data = self._transport.request("POST", "/upload", json_body=body, retry_safe=True)
return UploadTicket.model_validate(data)
def put_bytes(
self,
ticket: UploadTicket,
data: bytes,
*,
content_type: str | None = None,
) -> None:
"""PUT bytes directly to the presigned URL on object storage."""
self._transport.put_to_presigned_url(
ticket.upload_url,
data,
content_type or ticket.content_type,
content_length=len(data),
)
def upload_bytes(
self,
*,
filename: str,
data: bytes,
content_type: str | None = None,
) -> UploadTicket:
"""Convenience: presign + PUT in one call."""
ticket = self.create_ticket(filename=filename, content_type=content_type)
self.put_bytes(ticket, data, content_type=content_type)
return ticket
__all__ = ["UploadsResource"]