[PBE-1321] support campaign endpoints

stream_chat/async_chat/client.py

@@ -16,6 +16,9 @@
 )
 from urllib.parse import urlparse
 
+from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
+from stream_chat.types.segment import SegmentType, SegmentData, QuerySegmentsOptions, UpdateSegmentData
+
 if sys.version_info >= (3, 8):
     from typing import Literal
 else:
@@ -537,35 +540,37 @@ async def delete_role(self, name: str) -> StreamResponse:
     async def list_roles(self) -> StreamResponse:
         return await self.get("roles")
 
-    async def create_segment(self, segment: Dict) -> StreamResponse:
-        return await self.post("segments", data={"segment": segment})
+    async def create_segment(self, segment_type: SegmentType, segment_id: str, segment_name: str, data: SegmentData) -> StreamResponse:
+        return await self.post("segments", data={"type": segment_type.value, "id": segment_id, "name": segment_name, "data": data})
 
-    async def query_segments(self, **params: Any) -> StreamResponse:
-        return await self.get("segments", params={"payload": json.dumps(params)})
+    async def query_segments(self, filter_conditions: Dict, options: QuerySegmentsOptions) -> StreamResponse:
+        payload = {"filter": filter_conditions, **options}
+        return await self.post("segments/query", params=payload)
 
-    async def update_segment(self, segment_id: str, data: Dict) -> StreamResponse:
-        return await self.put(f"segments/{segment_id}", data={"segment": data})
+    async def update_segment(self, segment_id: str, data: UpdateSegmentData) -> StreamResponse:
+        return await self.put(f"segments/{segment_id}", data=data)
 
     async def delete_segment(self, segment_id: str) -> StreamResponse:
         return await self.delete(f"segments/{segment_id}")
 
-    async def create_campaign(self, campaign: Dict) -> StreamResponse:
-        return await self.post("campaigns", data={"campaign": campaign})
+    async def create_campaign(self, params: CampaignData) -> StreamResponse:
+        return await self.post("campaigns", data=params)
 
-    async def query_campaigns(self, **params: Any) -> StreamResponse:
-        return await self.get("campaigns", params={"payload": json.dumps(params)})
+    async def query_campaigns(self, filter_conditions: Dict[str, Any], options: QueryCampaignsOptions = None) -> StreamResponse:
+        payload = {"filter": filter_conditions, **options}
+        return await self.post("campaigns/query", params=payload)
 
-    async def update_campaign(self, campaign_id: str, data: Dict) -> StreamResponse:
-        return await self.put(f"campaigns/{campaign_id}", data={"campaign": data})
+    async def update_campaign(self, campaign_id: str, params: CampaignData) -> StreamResponse:
+        return await self.put(f"campaigns/{campaign_id}", data=params)
 
     async def delete_campaign(self, campaign_id: str, **options: Any) -> StreamResponse:
         return await self.delete(f"campaigns/{campaign_id}", params=options)
 
-    async def schedule_campaign(
-        self, campaign_id: str, scheduled_for: int = None
+    async def start_campaign(
+        self, campaign_id: str, scheduled_for: Optional[datetime.datetime] = None
     ) -> StreamResponse:
         return await self.patch(
-            f"campaigns/{campaign_id}/schedule", data={"scheduled_for": scheduled_for}
+            f"campaigns/{campaign_id}/start", data={"scheduled_for": scheduled_for}
         )
 
     async def query_recipients(self, **params: Any) -> StreamResponse:

stream_chat/base/campaign.py

@@ -0,0 +1,42 @@
+import abc
+from typing import Awaitable, Dict, List, Union, Optional
+
+from stream_chat.base.client import StreamChatInterface
+from stream_chat.types.campaign import CampaignData
+from stream_chat.types.stream_response import StreamResponse
+
+
+class CampaignInterface(abc.ABC):
+    def __init__(
+        self,
+        client: StreamChatInterface,
+        campaign_id: Optional[str] = None,
+        data: CampaignData = None,
+    ):
+        self.client = client
+        self.campaign_id = campaign_id
+        self.data = data
+
+    @abc.abstractmethod
+    def create(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def get(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def update(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def delete(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def start(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def stop(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass

stream_chat/base/client.py

@@ -5,7 +5,10 @@
 import hmac
 import os
 import sys
-from typing import Any, Awaitable, Dict, Iterable, List, TypeVar, Union
+from typing import Any, Awaitable, Dict, Iterable, List, TypeVar, Union, Optional
+
+from stream_chat.types.campaign import CampaignData, QueryCampaignsOptions
+from stream_chat.types.segment import SegmentType, SegmentData, QuerySegmentsOptions, UpdateSegmentData
 
 if sys.version_info >= (3, 8):
     from typing import Literal
@@ -18,6 +21,10 @@
 
 TChannel = TypeVar("TChannel")
 
+TSegment = TypeVar("TSegment")
+
+TCampaign = TypeVar("TCampaign")
+
 
 class StreamChatInterface(abc.ABC):
     def __init__(
@@ -585,7 +592,6 @@ def channel(
     ) -> TChannel:  # type: ignore[type-var]
         """
         Creates a channel object
-
         :param channel_type: the channel type
         :param channel_id: the id of the channel
         :param data: additional data, ie: {"members":[id1, id2, ...]}
@@ -918,9 +924,20 @@ def list_roles(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         pass
 
+    def segment(self, segment_type: SegmentType, segment_id: str, segment_name: str) -> TSegment:
+        """
+        Creates a channel object
+        :param segment_type: the segment type
+        :param segment_id: the id of the segment
+        :param segment_name: the name of the segment
+        :param data: additional data, ie: {"members":[id1, id2, ...]}
+        :return: Channel
+        """
+        pass
+
     @abc.abstractmethod
     def create_segment(
-        self, segment: Dict
+        self, segment_type: SegmentType, segment_id: str, segment_name: str, data: SegmentData
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Create a segment
@@ -929,7 +946,7 @@ def create_segment(
 
     @abc.abstractmethod
     def query_segments(
-        self, **params: Any
+        self, filter_conditions: Dict, options: QuerySegmentsOptions
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Query segments
@@ -938,7 +955,7 @@ def query_segments(
 
     @abc.abstractmethod
     def update_segment(
-        self, segment_id: str, data: Dict
+        self, segment_id: str, data: UpdateSegmentData
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Update a segment by id
@@ -954,9 +971,28 @@ def delete_segment(
         """
         pass
 
+    def campaign(self, campaign_id: Optional[str]):
+        """
+        Creates a channel object
+        :param channel_type: the channel type
+        :param channel_id: the id of the channel
+        :param data: additional data, ie: {"members":[id1, id2, ...]}
+        :return: Channel
+        """
+        pass
+
     @abc.abstractmethod
     def create_campaign(
-        self, campaign: Dict
+        self, campaign_id: Optional[str], data: Optional[CampaignData]
+    ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        """
+        Create a campaign
+        """
+        pass
+
+    @abc.abstractmethod
+    def get_campaign(
+        self, campaign_id: str
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Create a campaign
@@ -965,7 +1001,7 @@ def create_campaign(
 
     @abc.abstractmethod
     def query_campaigns(
-        self, **params: Any
+        self, filter_conditions: Dict[str, Any], options: QueryCampaignsOptions = None
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Query campaigns
@@ -974,7 +1010,7 @@ def query_campaigns(
 
     @abc.abstractmethod
     def update_campaign(
-        self, campaign_id: str, data: Dict
+        self, campaign_id: str, data: CampaignData
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Update a campaign
@@ -991,8 +1027,8 @@ def delete_campaign(
         pass
 
     @abc.abstractmethod
-    def schedule_campaign(
-        self, campaign_id: str, scheduled_for: int = None
+    def start_campaign(
+        self, campaign_id: str, scheduled_for: Optional[datetime.datetime] = None
     ) -> Union[StreamResponse, Awaitable[StreamResponse]]:
         """
         Schedule a campaign at given time

stream_chat/base/segment.py

@@ -0,0 +1,38 @@
+import abc
+from typing import Optional, Awaitable, Union
+
+from stream_chat.base.client import StreamChatInterface
+from stream_chat.types.segment import SegmentType, SegmentData, UpdateSegmentData
+from stream_chat.types.stream_response import StreamResponse
+
+
+class SegmentInterface(abc.ABC):
+    def __init__(
+        self,
+        client: StreamChatInterface,
+        segment_type: SegmentType,
+        segment_id: Optional[str] = None,
+        segment_name: Optional[str] = None,
+        data: Optional[SegmentData] = None,
+    ):
+        self.segment_type = segment_type
+        self.segment_id = segment_id
+        self.segment_name = segment_name
+        self.client = client
+        self.data = data
+
+    @abc.abstractmethod
+    def create(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def get(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def update(self, data: UpdateSegmentData) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass
+
+    @abc.abstractmethod
+    def delete(self) -> Union[StreamResponse, Awaitable[StreamResponse]]:
+        pass

stream_chat/campaign.py

@@ -0,0 +1,31 @@
+from stream_chat.base.campaign import CampaignInterface
+from stream_chat.types.stream_response import StreamResponse
+
+
+class Campaign(CampaignInterface):
+
+    def create(self) -> StreamResponse:
+        state = self.client.create_campaign(campaign_id=self.campaign_id, data=self.data)
+
+        if self.campaign_id is None and state.is_ok() and "campaign" in state:
+            self.campaign_id = state["campaign"]["id"]
+        return state
+
+    def get(self) -> StreamResponse:
+        return self.client.get_campaign(campaign_id=self.campaign_id)
+
+    def update(self, name: str = None, description: str = None) -> StreamResponse:
+        payload = {"name": name, "description": description}
+        return self.client.put(f"campaigns/{self.campaign_id}", data=payload)
+
+    def delete(self) -> StreamResponse:
+        return self.client.delete(f"campaigns/{self.campaign_id}")
+
+    def start(self) -> StreamResponse:
+        return self.client.get(f"campaigns/{self.campaign_id}")
+
+    def stop(self) -> StreamResponse:
+        return self.client.get(f"campaigns/{self.campaign_id}")
+
+    def query(self) -> StreamResponse:
+        return self.client.get(f"campaigns/{self.campaign_id}")

stream_chat/client.py

@@ -2,16 +2,20 @@
 import json
 import sys
 import warnings
-from typing import Any, Callable, Dict, Iterable, List, Union
+from typing import Any, Callable, Dict, Iterable, List, Union, Optional
 from urllib.parse import urlparse
 from urllib.request import Request, urlopen
 
+from stream_chat.campaign import Campaign
+from stream_chat.segment import Segment
+from stream_chat.types.campaign import QueryCampaignsOptions, CampaignData
+from stream_chat.types.segment import SegmentType, QuerySegmentsOptions, UpdateSegmentData, SegmentData
+
 if sys.version_info >= (3, 8):
     from typing import Literal
 else:
     from typing_extensions import Literal
 
-
 import requests
 
 from stream_chat.__pkg__ import __version__
@@ -518,35 +522,54 @@ def delete_role(self, name: str) -> StreamResponse:
     def list_roles(self) -> StreamResponse:
         return self.get("roles")
 
-    def create_segment(self, segment: Dict) -> StreamResponse:
-        return self.post("segments", data={"segment": segment})
+    def segment(self, segment_type: SegmentType, segment_id: Optional[str] = None, segment_name: Optional[str] = None,
+                data: Optional[SegmentData] = None) -> Segment:
+        return Segment(client=self, segment_type=segment_type, segment_id=segment_id, segment_name=segment_name,
+                       data=data)
+
+    def create_segment(self, segment_type: SegmentType, segment_id: Optional[str] = None,
+                       segment_name: Optional[str] = None, data: Dict = None) -> StreamResponse:
+        return self.post("segments",
+                         data={"type": segment_type.value, "id": segment_id, "name": segment_name, "data": data})
 
-    def query_segments(self, **params: Any) -> StreamResponse:
-        return self.get("segments", params={"payload": json.dumps(params)})
+    def query_segments(self, filter_conditions: Dict, options: QuerySegmentsOptions) -> StreamResponse:
+        payload = {"filter": filter_conditions, **options}
+        return self.post("segments/query", data=payload)
 
-    def update_segment(self, segment_id: str, data: Dict) -> StreamResponse:
-        return self.put(f"segments/{segment_id}", data={"segment": data})
+    def update_segment(self, segment_id: str, data: UpdateSegmentData) -> StreamResponse:
+        return self.put(f"segments/{segment_id}", data=data)
 
     def delete_segment(self, segment_id: str) -> StreamResponse:
         return self.delete(f"segments/{segment_id}")
 
-    def create_campaign(self, campaign: Dict) -> StreamResponse:
-        return self.post("campaigns", data={"campaign": campaign})
+    def campaign(self, campaign_id: Optional[str] = None, data: CampaignData = None) -> Campaign:
+        return Campaign(client=self, campaign_id=campaign_id, data=data)
+
+    def create_campaign(self, campaign_id: Optional[str] = None, data: CampaignData = None) -> StreamResponse:
+        payload = {"id": campaign_id}
+        if data is not None:
+            payload.update(data)
+        return self.post("campaigns", data=payload)
+
+    def get_campaign(self, campaign_id: str) -> StreamResponse:
+        return self.get(f"campaigns/{campaign_id}")
 
-    def query_campaigns(self, **params: Any) -> StreamResponse:
-        return self.get("campaigns", params={"payload": json.dumps(params)})
+    def query_campaigns(self, filter_conditions: Dict[str, Any],
+                        options: QueryCampaignsOptions = None) -> StreamResponse:
+        payload = {"filter": filter_conditions, **options}
+        return self.post("campaigns/query", data=payload)
 
-    def update_campaign(self, campaign_id: str, data: Dict) -> StreamResponse:
-        return self.put(f"campaigns/{campaign_id}", data={"campaign": data})
+    def update_campaign(self, campaign_id: str, params: CampaignData) -> StreamResponse:
+        return self.put(f"campaigns/{campaign_id}", data=params)
 
     def delete_campaign(self, campaign_id: str, **options: Any) -> StreamResponse:
         return self.delete(f"campaigns/{campaign_id}", params=options)
 
-    def schedule_campaign(
-        self, campaign_id: str, scheduled_for: int = None
+    def start_campaign(
+            self, campaign_id: str, scheduled_for: Optional[datetime.datetime] = None
     ) -> StreamResponse:
         return self.patch(
-            f"campaigns/{campaign_id}/schedule", data={"scheduled_for": scheduled_for}
+            f"campaigns/{campaign_id}/start", data={"scheduled_for": scheduled_for}
         )
 
     def stop_campaign(self, campaign_id: str) -> StreamResponse: