Skip to content

Commit feb2d55

Browse files
alievaliev
authored
Merge branch 'main' into fix/stop-drain-on-subscribe
2 parents 032e3de + 97bb477 commit feb2d55

7 files changed

Lines changed: 359 additions & 37 deletions

File tree

‎getstream/base.py‎

Lines changed: 74 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
import json
2+
import mimetypes
3+
import os
24
import time
35
import uuid
46
import asyncio
5-
from typing import Any, Dict, Optional, Type, cast, get_origin
7+
from typing import Any, Dict, List, Optional, Tuple, Type, cast, get_origin
68

79
from getstream.models import APIError
810
from getstream.rate_limit import extract_rate_limit
@@ -25,6 +27,11 @@
2527
import ijson
2628

2729

30+
def _read_file_bytes(file_path: str) -> bytes:
31+
with open(file_path, "rb") as f:
32+
return f.read()
33+
34+
2835
def _strip_none(obj):
2936
"""Recursively remove None values from dicts so unset optional fields
3037
are omitted from the JSON body instead of being sent as null."""
@@ -305,6 +312,39 @@ def delete(
305312
data_type=data_type,
306313
)
307314

315+
def _upload_multipart(
316+
self,
317+
path: str,
318+
data_type: Type[T],
319+
file_path: str,
320+
*,
321+
path_params: Optional[Dict[str, str]] = None,
322+
query_params: Optional[Dict[str, str]] = None,
323+
form_fields: Optional[List[Tuple[str, str]]] = None,
324+
) -> StreamResponse[T]:
325+
"""Send a multipart/form-data upload request, matching Go/PHP SDK behavior."""
326+
file_name = os.path.basename(file_path)
327+
content_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
328+
with open(file_path, "rb") as f:
329+
file_content = f.read()
330+
331+
files = {"file": (file_name, file_content, content_type)}
332+
data: Dict[str, str] = {}
333+
for field_name, field_value in form_fields or []:
334+
data[field_name] = field_value
335+
336+
kwargs: Dict[str, Any] = {"files": files}
337+
if data:
338+
kwargs["data"] = data
339+
340+
return self._request_sync(
341+
"POST",
342+
path,
343+
query_params=query_params,
344+
kwargs=kwargs | {"path_params": path_params},
345+
data_type=data_type,
346+
)
347+
308348
def close(self):
309349
"""
310350
Close HTTPX client.
@@ -345,6 +385,39 @@ async def aclose(self):
345385
"""Close HTTPX async client (closes pools/keep-alives)."""
346386
await self.client.aclose()
347387

388+
async def _upload_multipart(
389+
self,
390+
path: str,
391+
data_type: Type[T],
392+
file_path: str,
393+
*,
394+
path_params: Optional[Dict[str, str]] = None,
395+
query_params: Optional[Dict[str, str]] = None,
396+
form_fields: Optional[List[Tuple[str, str]]] = None,
397+
) -> StreamResponse[T]:
398+
"""Send a multipart/form-data upload request, matching Go/PHP SDK behavior."""
399+
file_name = os.path.basename(file_path)
400+
content_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
401+
402+
file_content = await asyncio.to_thread(_read_file_bytes, file_path)
403+
404+
files = {"file": (file_name, file_content, content_type)}
405+
data: Dict[str, str] = {}
406+
for field_name, field_value in form_fields or []:
407+
data[field_name] = field_value
408+
409+
kwargs: Dict[str, Any] = {"files": files}
410+
if data:
411+
kwargs["data"] = data
412+
413+
return await self._request_async(
414+
"POST",
415+
path,
416+
query_params=query_params,
417+
kwargs=kwargs | {"path_params": path_params},
418+
data_type=data_type,
419+
)
420+
348421
def _endpoint_name(self, path: str) -> str:
349422
op = getattr(self, "_operation_name", None)
350423
return op or current_operation(self._normalize_endpoint_from_path(path)) or ""

‎getstream/chat/async_client.py‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,16 @@
1+
import json
2+
from typing import List, Optional
3+
14
from getstream.chat.async_channel import Channel
25
from getstream.chat.async_rest_client import ChatRestClient
6+
from getstream.common import telemetry
7+
from getstream.models import (
8+
ImageSize,
9+
OnlyUserID,
10+
UploadChannelFileResponse,
11+
UploadChannelResponse,
12+
)
13+
from getstream.stream_response import StreamResponse
314

415

516
class ChatClient(ChatRestClient):
@@ -15,3 +26,60 @@ def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=No
1526

1627
def channel(self, call_type: str, id: str) -> Channel:
1728
return Channel(self, call_type, id)
29+
30+
@telemetry.operation_name("getstream.api.chat.upload_channel_file")
31+
async def upload_channel_file(
32+
self,
33+
type: str,
34+
id: str,
35+
file: str,
36+
user: Optional[OnlyUserID] = None,
37+
) -> StreamResponse[UploadChannelFileResponse]:
38+
form_fields = []
39+
if user is not None:
40+
form_fields.append(("user", json.dumps(user.to_dict())))
41+
return await self._upload_multipart(
42+
"/api/v2/chat/channels/{type}/{id}/file",
43+
UploadChannelFileResponse,
44+
file,
45+
path_params={"type": type, "id": id},
46+
form_fields=form_fields,
47+
)
48+
49+
@telemetry.operation_name("getstream.api.chat.upload_channel_image")
50+
async def upload_channel_image(
51+
self,
52+
channel_type: Optional[str] = None,
53+
id: Optional[str] = None,
54+
file: Optional[str] = None,
55+
upload_sizes: Optional[List[ImageSize]] = None,
56+
user: Optional[OnlyUserID] = None,
57+
**kwargs,
58+
) -> StreamResponse[UploadChannelResponse]:
59+
# Backward compatibility for generated wrappers passing `type=...`.
60+
if channel_type is None:
61+
channel_type = kwargs.pop("type", None)
62+
if kwargs:
63+
raise TypeError(f"Unexpected keyword arguments: {', '.join(kwargs.keys())}")
64+
if channel_type is None:
65+
raise TypeError(
66+
"upload_channel_image() missing required argument: 'channel_type'"
67+
)
68+
if id is None:
69+
raise TypeError("upload_channel_image() missing required argument: 'id'")
70+
if file is None:
71+
raise TypeError("upload_channel_image() missing required argument: 'file'")
72+
form_fields = []
73+
if user is not None:
74+
form_fields.append(("user", json.dumps(user.to_dict())))
75+
if upload_sizes is not None:
76+
form_fields.append(
77+
("upload_sizes", json.dumps([s.to_dict() for s in upload_sizes]))
78+
)
79+
return await self._upload_multipart(
80+
"/api/v2/chat/channels/{type}/{id}/image",
81+
UploadChannelResponse,
82+
file,
83+
path_params={"type": channel_type, "id": id},
84+
form_fields=form_fields,
85+
)

‎getstream/chat/client.py‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,16 @@
1+
import json
2+
from typing import List, Optional
3+
14
from getstream.chat.channel import Channel
25
from getstream.chat.rest_client import ChatRestClient
6+
from getstream.common import telemetry
7+
from getstream.models import (
8+
ImageSize,
9+
OnlyUserID,
10+
UploadChannelFileResponse,
11+
UploadChannelResponse,
12+
)
13+
from getstream.stream_response import StreamResponse
314

415

516
class ChatClient(ChatRestClient):
@@ -15,3 +26,60 @@ def __init__(self, api_key: str, base_url, token, timeout, stream, user_agent=No
1526

1627
def channel(self, call_type: str, id: str) -> Channel:
1728
return Channel(self, call_type, id)
29+
30+
@telemetry.operation_name("getstream.api.chat.upload_channel_file")
31+
def upload_channel_file(
32+
self,
33+
type: str,
34+
id: str,
35+
file: str,
36+
user: Optional[OnlyUserID] = None,
37+
) -> StreamResponse[UploadChannelFileResponse]:
38+
form_fields = []
39+
if user is not None:
40+
form_fields.append(("user", json.dumps(user.to_dict())))
41+
return self._upload_multipart(
42+
"/api/v2/chat/channels/{type}/{id}/file",
43+
UploadChannelFileResponse,
44+
file,
45+
path_params={"type": type, "id": id},
46+
form_fields=form_fields,
47+
)
48+
49+
@telemetry.operation_name("getstream.api.chat.upload_channel_image")
50+
def upload_channel_image(
51+
self,
52+
channel_type: Optional[str] = None,
53+
id: Optional[str] = None,
54+
file: Optional[str] = None,
55+
upload_sizes: Optional[List[ImageSize]] = None,
56+
user: Optional[OnlyUserID] = None,
57+
**kwargs,
58+
) -> StreamResponse[UploadChannelResponse]:
59+
# Backward compatibility for generated wrappers passing `type=...`.
60+
if channel_type is None:
61+
channel_type = kwargs.pop("type", None)
62+
if kwargs:
63+
raise TypeError(f"Unexpected keyword arguments: {', '.join(kwargs.keys())}")
64+
if channel_type is None:
65+
raise TypeError(
66+
"upload_channel_image() missing required argument: 'channel_type'"
67+
)
68+
if id is None:
69+
raise TypeError("upload_channel_image() missing required argument: 'id'")
70+
if file is None:
71+
raise TypeError("upload_channel_image() missing required argument: 'file'")
72+
form_fields = []
73+
if user is not None:
74+
form_fields.append(("user", json.dumps(user.to_dict())))
75+
if upload_sizes is not None:
76+
form_fields.append(
77+
("upload_sizes", json.dumps([s.to_dict() for s in upload_sizes]))
78+
)
79+
return self._upload_multipart(
80+
"/api/v2/chat/channels/{type}/{id}/image",
81+
UploadChannelResponse,
82+
file,
83+
path_params={"type": channel_type, "id": id},
84+
form_fields=form_fields,
85+
)

‎getstream/common/async_client.py‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,15 @@
1+
import json
2+
from typing import List, Optional
3+
4+
from getstream.common import telemetry
15
from getstream.common.async_rest_client import CommonRestClient
6+
from getstream.models import (
7+
FileUploadResponse,
8+
ImageSize,
9+
ImageUploadResponse,
10+
OnlyUserID,
11+
)
12+
from getstream.stream_response import StreamResponse
213

314

415
class CommonClient(CommonRestClient):
@@ -10,3 +21,38 @@ def __init__(self, api_key: str, base_url, token, timeout, user_agent=None):
1021
timeout=timeout,
1122
user_agent=user_agent,
1223
)
24+
25+
@telemetry.operation_name("getstream.api.common.upload_file")
26+
async def upload_file(
27+
self, file: str, user: Optional[OnlyUserID] = None
28+
) -> StreamResponse[FileUploadResponse]:
29+
form_fields = []
30+
if user is not None:
31+
form_fields.append(("user", json.dumps(user.to_dict())))
32+
return await self._upload_multipart(
33+
"/api/v2/uploads/file",
34+
FileUploadResponse,
35+
file,
36+
form_fields=form_fields,
37+
)
38+
39+
@telemetry.operation_name("getstream.api.common.upload_image")
40+
async def upload_image(
41+
self,
42+
file: str,
43+
upload_sizes: Optional[List[ImageSize]] = None,
44+
user: Optional[OnlyUserID] = None,
45+
) -> StreamResponse[ImageUploadResponse]:
46+
form_fields = []
47+
if user is not None:
48+
form_fields.append(("user", json.dumps(user.to_dict())))
49+
if upload_sizes is not None:
50+
form_fields.append(
51+
("upload_sizes", json.dumps([s.to_dict() for s in upload_sizes]))
52+
)
53+
return await self._upload_multipart(
54+
"/api/v2/uploads/image",
55+
ImageUploadResponse,
56+
file,
57+
form_fields=form_fields,
58+
)

‎getstream/common/client.py‎

Lines changed: 46 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,15 @@
1+
import json
2+
from typing import List, Optional
3+
4+
from getstream.common import telemetry
15
from getstream.common.rest_client import CommonRestClient
6+
from getstream.models import (
7+
FileUploadResponse,
8+
ImageSize,
9+
ImageUploadResponse,
10+
OnlyUserID,
11+
)
12+
from getstream.stream_response import StreamResponse
213

314

415
class CommonClient(CommonRestClient):
@@ -10,3 +21,38 @@ def __init__(self, api_key: str, base_url, token, timeout, user_agent=None):
1021
timeout=timeout,
1122
user_agent=user_agent,
1223
)
24+
25+
@telemetry.operation_name("getstream.api.common.upload_file")
26+
def upload_file(
27+
self, file: str, user: Optional[OnlyUserID] = None
28+
) -> StreamResponse[FileUploadResponse]:
29+
form_fields = []
30+
if user is not None:
31+
form_fields.append(("user", json.dumps(user.to_dict())))
32+
return self._upload_multipart(
33+
"/api/v2/uploads/file",
34+
FileUploadResponse,
35+
file,
36+
form_fields=form_fields,
37+
)
38+
39+
@telemetry.operation_name("getstream.api.common.upload_image")
40+
def upload_image(
41+
self,
42+
file: str,
43+
upload_sizes: Optional[List[ImageSize]] = None,
44+
user: Optional[OnlyUserID] = None,
45+
) -> StreamResponse[ImageUploadResponse]:
46+
form_fields = []
47+
if user is not None:
48+
form_fields.append(("user", json.dumps(user.to_dict())))
49+
if upload_sizes is not None:
50+
form_fields.append(
51+
("upload_sizes", json.dumps([s.to_dict() for s in upload_sizes]))
52+
)
53+
return self._upload_multipart(
54+
"/api/v2/uploads/image",
55+
ImageUploadResponse,
56+
file,
57+
form_fields=form_fields,
58+
)

0 commit comments

Comments
 (0)