22# Copyright (c) 2022 Airbyte, Inc., all rights reserved.
33#
44
5- from __future__ import annotations
65
76from abc import ABC
8- from typing import Any , Iterable , List , Mapping , MutableMapping , Optional , Type
7+ from typing import Any , Iterable , Mapping , MutableMapping , Optional , Type
98
109import requests
1110from airbyte_cdk .models import SyncMode
1211from airbyte_cdk .sources .streams .http import HttpStream
1312
13+ ASANA_ERRORS_MAPPING = {
14+ 402 : "This stream is available to premium organizations and workspaces only" ,
15+ 403 : "Missing permissions to consume this stream enough permissions" ,
16+ 404 : "The object specified by the request does not exist" ,
17+ 451 : "This request was blocked for legal reasons" ,
18+ }
19+
1420
1521class AsanaStream (HttpStream , ABC ):
1622 url_base = "https://app.asana.com/api/1.0/"
17-
1823 primary_key = "gid"
19-
2024 # Asana pagination could be from 1 to 100.
2125 page_size = 100
26+ raise_on_http_errors = True
27+
28+ @property
29+ def AsanaStreamType (self ) -> Type :
30+ return self .__class__
31+
32+ def should_retry (self , response : requests .Response ) -> bool :
33+ if response .status_code in ASANA_ERRORS_MAPPING .keys ():
34+ self .logger .error (
35+ f"Skipping stream { self .name } { ASANA_ERRORS_MAPPING .get (response .status_code )} { response .text }
36+ )
37+ setattr (self , "raise_on_http_errors" , False )
38+ return False
39+ return super ().should_retry (response )
2240
2341 def backoff_time (self , response : requests .Response ) -> Optional [int ]:
2442 delay_time = response .headers .get ("Retry-After" )
@@ -31,17 +49,11 @@ def next_page_token(self, response: requests.Response) -> Optional[Mapping[str,
3149 if next_page :
3250 return {"offset" : next_page ["offset" ]}
3351
34- def request_params (
35- self , stream_state : Mapping [str , Any ], stream_slice : Mapping [str , any ] = None , next_page_token : Mapping [str , Any ] = None
36- ) -> MutableMapping [str , Any ]:
37-
52+ def request_params (self , next_page_token : Mapping [str , Any ] = None , ** kwargs ) -> MutableMapping [str , Any ]:
3853 params = {"limit" : self .page_size }
39-
4054 params .update (self .get_opt_fields ())
41-
4255 if next_page_token :
4356 params .update (next_page_token )
44-
4557 return params
4658
4759 def get_opt_fields (self ) -> MutableMapping [str , str ]:
@@ -81,7 +93,7 @@ def parse_response(self, response: requests.Response, **kwargs) -> Iterable[Mapp
8193 response_json = response .json ()
8294 yield from response_json .get ("data" , []) # Asana puts records in a container array "data"
8395
84- def read_slices_from_records (self , stream_class : Type [ AsanaStream ] , slice_field : str ) -> Iterable [Optional [Mapping [str , Any ]]]:
96+ def read_slices_from_records (self , stream_class : AsanaStreamType , slice_field : str ) -> Iterable [Optional [Mapping [str , Any ]]]:
8597 """
8698 General function for getting parent stream (which should be passed through `stream_class`) slice.
8799 Generates dicts with `gid` of parent streams.
@@ -100,9 +112,7 @@ class WorkspaceRelatedStream(AsanaStream, ABC):
100112 into the path or will pass it as a request parameter.
101113 """
102114
103- def stream_slices (
104- self , sync_mode : SyncMode , cursor_field : List [str ] = None , stream_state : Mapping [str , Any ] = None
105- ) -> Iterable [Optional [Mapping [str , Any ]]]:
115+ def stream_slices (self , ** kwargs ) -> Iterable [Optional [Mapping [str , Any ]]]:
106116 workspaces_stream = Workspaces (authenticator = self .authenticator )
107117 for workspace in workspaces_stream .read_records (sync_mode = SyncMode .full_refresh ):
108118 yield {"workspace_gid" : workspace ["gid" ]}
@@ -114,10 +124,8 @@ class WorkspaceRequestParamsRelatedStream(WorkspaceRelatedStream, ABC):
114124 So this is basically the whole point of this class - to pass `workspace` as request argument.
115125 """
116126
117- def request_params (
118- self , stream_state : Mapping [str , Any ], stream_slice : Mapping [str , Any ] = None , next_page_token : Mapping [str , Any ] = None
119- ) -> MutableMapping [str , Any ]:
120- params = super ().request_params (stream_state = stream_state , stream_slice = stream_slice , next_page_token = next_page_token )
127+ def request_params (self , stream_slice : Mapping [str , Any ] = None , ** kwargs ) -> MutableMapping [str , Any ]:
128+ params = super ().request_params (** kwargs )
121129 params ["workspace" ] = stream_slice ["workspace_gid" ]
122130 return params
123131
@@ -128,9 +136,7 @@ class ProjectRelatedStream(AsanaStream, ABC):
128136 argument in request.
129137 """
130138
131- def stream_slices (
132- self , sync_mode : SyncMode , cursor_field : List [str ] = None , stream_state : Mapping [str , Any ] = None
133- ) -> Iterable [Optional [Mapping [str , Any ]]]:
139+ def stream_slices (self , ** kwargs ) -> Iterable [Optional [Mapping [str , Any ]]]:
134140 yield from self .read_slices_from_records (stream_class = Projects , slice_field = "project_gid" )
135141
136142
@@ -158,9 +164,7 @@ def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
158164 task_gid = stream_slice ["task_gid" ]
159165 return f"tasks/{ task_gid }
160166
161- def stream_slices (
162- self , sync_mode : SyncMode , cursor_field : List [str ] = None , stream_state : Mapping [str , Any ] = None
163- ) -> Iterable [Optional [Mapping [str , Any ]]]:
167+ def stream_slices (self , ** kwargs ) -> Iterable [Optional [Mapping [str , Any ]]]:
164168 yield from self .read_slices_from_records (stream_class = Tasks , slice_field = "task_gid" )
165169
166170
@@ -173,10 +177,8 @@ class Tasks(ProjectRelatedStream):
173177 def path (self , ** kwargs ) -> str :
174178 return "tasks"
175179
176- def request_params (
177- self , stream_state : Mapping [str , Any ], stream_slice : Mapping [str , Any ] = None , next_page_token : Mapping [str , Any ] = None
178- ) -> MutableMapping [str , Any ]:
179- params = super ().request_params (stream_state = stream_state , stream_slice = stream_slice , next_page_token = next_page_token )
180+ def request_params (self , stream_slice : Mapping [str , Any ] = None , ** kwargs ) -> MutableMapping [str , Any ]:
181+ params = super ().request_params (stream_slice = stream_slice , ** kwargs )
180182 params ["project" ] = stream_slice ["project_gid" ]
181183 return params
182184
@@ -202,9 +204,7 @@ def path(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> str:
202204 team_gid = stream_slice ["team_gid" ]
203205 return f"teams/{ team_gid }
204206
205- def stream_slices (
206- self , sync_mode : SyncMode , cursor_field : List [str ] = None , stream_state : Mapping [str , Any ] = None
207- ) -> Iterable [Optional [Mapping [str , Any ]]]:
207+ def stream_slices (self , ** kwargs ) -> Iterable [Optional [Mapping [str , Any ]]]:
208208 yield from self .read_slices_from_records (stream_class = Teams , slice_field = "team_gid" )
209209
210210
0 commit comments