|
28 | 28 |
|
29 | 29 | import zulip
|
30 | 30 | from bs4 import BeautifulSoup
|
31 |
| -from typing_extensions import TypedDict |
| 31 | +from typing_extensions import Literal, TypedDict |
32 | 32 |
|
33 | 33 | from zulipterminal import unicode_emojis
|
34 | 34 | from zulipterminal.api_types import (
|
@@ -1297,6 +1297,53 @@ def _get_all_stream_ids(self) -> List[int]:
|
1297 | 1297 | id_list.extend(stream_id for stream_id in self._never_subscribed_streams)
|
1298 | 1298 | return id_list
|
1299 | 1299 |
|
| 1300 | + STREAM_KEYS = Literal[ |
| 1301 | + "stream_id", |
| 1302 | + "name", |
| 1303 | + "description", |
| 1304 | + "rendered_description", |
| 1305 | + "date_created", |
| 1306 | + "invite_only", |
| 1307 | + "subscribers", |
| 1308 | + "is_announcement_only", |
| 1309 | + "stream_post_policy", |
| 1310 | + "is_web_public", |
| 1311 | + "message_retention_days", |
| 1312 | + "history_public_to_subscribers", |
| 1313 | + "first_message_id", |
| 1314 | + "stream_weekly_traffic", |
| 1315 | + ] |
| 1316 | + SUBSCRIPTION_KEYS = Literal[ |
| 1317 | + STREAM_KEYS, |
| 1318 | + "desktop_notifications", |
| 1319 | + "email_notifications", |
| 1320 | + "wildcard_mentions_notify", |
| 1321 | + "push_notifications", |
| 1322 | + "audible_notifications", |
| 1323 | + "pin_to_top", |
| 1324 | + "email_address", |
| 1325 | + "is_muted", |
| 1326 | + "color", |
| 1327 | + ] |
| 1328 | + |
| 1329 | + def stream_property( |
| 1330 | + self, stream_id: int, property: STREAM_KEYS |
| 1331 | + ) -> Optional[Union[int, str, bool, List[int]]]: |
| 1332 | + return self._get_stream_from_id(stream_id)[property] |
| 1333 | + |
| 1334 | + def subscription_property( |
| 1335 | + self, stream_id: int, property: SUBSCRIPTION_KEYS |
| 1336 | + ) -> Optional[Union[int, str, bool, List[int]]]: |
| 1337 | + subscription = self._get_stream_from_id(stream_id) |
| 1338 | + if property in subscription: |
| 1339 | + subscription = cast(Subscription, subscription) |
| 1340 | + return subscription[property] |
| 1341 | + else: |
| 1342 | + # stream_id is not a subscribed stream. |
| 1343 | + raise RuntimeError( |
| 1344 | + f"Stream with id={stream_id} does not have '{property}' property!" |
| 1345 | + ) |
| 1346 | + |
1300 | 1347 | def _subscribe_to_streams(self, subscriptions: List[Subscription]) -> None:
|
1301 | 1348 | def make_reduced_stream_data(stream: Subscription) -> StreamData:
|
1302 | 1349 | # stream_id has been changed to id.
|
|
0 commit comments