-
import marshmallow_dataclass
+
from pydantic.dataclasses import dataclass
from typing import Any, Iterable, List, Mapping
-
class BadResponse(Exception):
-
super().__init__("Bad response")
-
class GetFailed(Exception):
-
def __init__(self, status, reason, content):
-
super().__init__(f"{status} {reason}\n{content})")
+
from ideascale_importer import utils
class ExcludeUnknownFields:
-
unknown = marshmallow.EXCLUDE
class Campaign(ExcludeUnknownFields):
Represents a campaign from IdeaScale.
class CampaignGroup(ExcludeUnknownFields):
Represents a campaign group from IdeaScale.
campaigns: List[Campaign]
class IdeaAuthorInfo(ExcludeUnknownFields):
Represents an author info from IdeaScale.
class Idea(ExcludeUnknownFields):
Represents an idea from IdeaScale.
return list(map(lambda c: c.name, self.contributors))
class Stage(ExcludeUnknownFields):
Represents a stage from IdeaScale.
class Funnel(ExcludeUnknownFields):
Represents a funnel from IdeaScale.
-
CampaignSchema = marshmallow_dataclass.class_schema(Campaign)
-
CampaignGroupSchema = marshmallow_dataclass.class_schema(CampaignGroup)
-
IdeaAuthorInfoSchema = marshmallow_dataclass.class_schema(IdeaAuthorInfo)
-
IdeaSchema = marshmallow_dataclass.class_schema(Idea)
-
StageSchema = marshmallow_dataclass.class_schema(Stage)
-
FunnelSchema = marshmallow_dataclass.class_schema(Funnel)
-
class RequestProgressObserver:
-
Observer used for displaying IdeaScale client's requests progresses.
-
self.inflight_requests = {}
-
self.progress = rich.progress.Progress(
-
rich.progress.TextColumn("{task.description}"),
-
rich.progress.DownloadColumn(),
-
rich.progress.TransferSpeedColumn(),
-
rich.progress.SpinnerColumn(),
-
def request_start(self, req_id: int, method: str, url: str):
-
self.inflight_requests[req_id] = [self.progress.add_task(f"({req_id}) {method} {url}", total=None), 0]
-
def request_progress(self, req_id: int, total_bytes_received: int):
-
self.inflight_requests[req_id][1] = total_bytes_received
-
self.progress.update(self.inflight_requests[req_id][0], completed=total_bytes_received)
-
def request_end(self, req_id):
-
self.progress.update(self.inflight_requests[req_id][0], total=self.inflight_requests[req_id][1])
-
self.progress.__enter__()
-
def __exit__(self, *args):
-
self.progress.__exit__(*args)
-
for [task_id, _] in self.inflight_requests.values():
-
self.progress.remove_task(task_id)
-
self.inflight_requests.clear()
def __init__(self, api_token: str):
self.api_token = api_token
-
self.request_counter = 0
-
self.request_progress_observer = RequestProgressObserver()
+
self.inner = utils.JsonHttpClient(Client.API_URL)
async def campaigns(self, group_id: int) -> List[Campaign]:
assert isinstance(group, dict)
-
campaigns.extend(CampaignSchema().load(group["campaigns"], many=True) or [])
+
for c in group["campaigns"]:
+
pydantic.tools.parse_obj_as(Campaign, c)
+
campaigns.extend(group_campaigns)
res = await self._get("/v1/campaigns/groups")
-
return CampaignGroupSchema().load(res, many=True) or []
+
campaign_groups: List[CampaignGroup] = []
+
pydantic.tools.parse_obj_as(CampaignGroup, cg)
async def campaign_ideas(self, campaign_id: int) -> List[Idea]:
Gets all ideas from the campaign with the given id.
res = await self._get(f"/v1/campaigns/{campaign_id}/ideas")
-
return IdeaSchema().load(res, many=True) or []
+
pydantic.tools.parse_obj_as(Idea, i)
async def stage_ideas(self, stage_id: int, page_size: int = 50, request_workers_count: int = 10) -> List[Idea]:
res = await self._get(f"/v1/stages/{stage_id}/ideas/{p}/{page_size}")
-
res_ideas = IdeaSchema().load(res, many=True) or []
+
res_ideas: List[Idea] = []
+
pydantic.tools.parse_obj_as(Idea, i)
d.ideas.extend(res_ideas)
Gets the funnel with the given id.
-
funnel = FunnelSchema().load(await self._get(f"/v1/funnels/{funnel_id}"))
-
if isinstance(funnel, Funnel):
