55
66import logging
77from collections import Counter
8+ from json import JSONDecodeError
89from typing import Any , List , Mapping , Tuple , Union
910
1011import requests
1617
1718
1819class SourceNetsuite (AbstractSource ):
20+
21+ logger : logging .Logger = logging .getLogger ("airbyte" )
22+
1923 def auth (self , config : Mapping [str , Any ]) -> OAuth1 :
2024 return OAuth1 (
2125 client_key = config ["consumer_key" ],
@@ -50,7 +54,7 @@ def check_connection(self, logger, config: Mapping[str, Any]) -> Tuple[bool, Any
5054 # check connectivity to all provided `object_types`
5155 for object in object_types :
5256 try :
53- response = session .get (url = base_url + RECORD_PATH + object , params = {"limit" : 1 })
57+ response = session .get (url = base_url + RECORD_PATH + object . lower () , params = {"limit" : 1 })
5458 response .raise_for_status ()
5559 return True , None
5660 except requests .exceptions .HTTPError as e :
@@ -67,11 +71,29 @@ def check_connection(self, logger, config: Mapping[str, Any]) -> Tuple[bool, Any
6771 return False , e
6872
6973 def get_schemas (self , object_names : Union [List [str ], str ], session : requests .Session , metadata_url : str ) -> Mapping [str , Any ]:
70- # fetch schemas
71- if isinstance (object_names , list ):
72- return {object_name : session .get (metadata_url + object_name , headers = SCHEMA_HEADERS ).json () for object_name in object_names }
73- elif isinstance (object_names , str ):
74- return {object_names : session .get (metadata_url + object_names , headers = SCHEMA_HEADERS ).json ()}
74+ """
75+ Handles multivariance of object_names type input and fetches the schema for each object type provided.
76+ """
77+ try :
78+ if isinstance (object_names , list ):
79+ schemas = {}
80+ for object_name in object_names :
81+ schemas .update (** self .fetch_schema (object_name , session , metadata_url ))
82+ return schemas
83+ elif isinstance (object_names , str ):
84+ return self .fetch_schema (object_names , session , metadata_url )
85+ else :
86+ raise NotImplementedError (
87+ f"Object Types has unknown structure, should be either `dict` or `str`, actual input: { object_names }
88+ )
89+ except JSONDecodeError as e :
90+ self .logger .error (f"Unexpected output while fetching the object schema. Full error: { e .__repr__ ()} )
91+
92+ def fetch_schema (self , object_name : str , session : requests .Session , metadata_url : str ) -> Mapping [str , Any ]:
93+ """
94+ Calls the API for specific object type and returns schema as a dict.
95+ """
96+ return {object_name .lower (): session .get (metadata_url + object_name , headers = SCHEMA_HEADERS ).json ()}
7597
7698 def generate_stream (
7799 self ,
@@ -83,35 +105,40 @@ def generate_stream(
83105 base_url : str ,
84106 start_datetime : str ,
85107 window_in_days : int ,
108+ max_retry : int = 3 ,
86109 ) -> Union [NetsuiteStream , IncrementalNetsuiteStream , CustomIncrementalNetsuiteStream ]:
87110
88- logger : logging .Logger = (logging .Logger ,)
89-
90111 input_args = {
91112 "auth" : auth ,
92113 "object_name" : object_name ,
93114 "base_url" : base_url ,
94115 "start_datetime" : start_datetime ,
95116 "window_in_days" : window_in_days ,
96117 }
97- try :
98- schema = schemas [object_name ]
99- schema_props = schema ["properties" ]
100- if schema_props :
101- if INCREMENTAL_CURSOR in schema_props .keys ():
102- return IncrementalNetsuiteStream (** input_args )
103- elif CUSTOM_INCREMENTAL_CURSOR in schema_props .keys ():
104- return CustomIncrementalNetsuiteStream (** input_args )
105- else :
106- # all other streams are full_refresh
107- return NetsuiteStream (** input_args )
108- except KeyError :
109- logger .warn (f"Object `{ object_name } )
110- # somethimes object metadata returns data with missing `properties` key,
111- # we should try to fetch metadata again to that object
112- schemas = self .get_schemas (object_name , session , metadata_url )
113- input_args .update (** {"session" : session , "metadata_url" : metadata_url , "schemas" : schemas })
114- return self .generate_stream (** input_args )
118+
119+ schema = schemas [object_name ]
120+ schema_props = schema .get ("properties" )
121+ if schema_props :
122+ if INCREMENTAL_CURSOR in schema_props .keys ():
123+ return IncrementalNetsuiteStream (** input_args )
124+ elif CUSTOM_INCREMENTAL_CURSOR in schema_props .keys ():
125+ return CustomIncrementalNetsuiteStream (** input_args )
126+ else :
127+ # all other streams are full_refresh
128+ return NetsuiteStream (** input_args )
129+ else :
130+ retry_attempt = 1
131+ while retry_attempt <= max_retry :
132+ self .logger .warn (f"Object `{ object_name } { retry_attempt } { max_retry } )
133+ # somethimes object metadata returns data with missing `properties` key,
134+ # we should try to fetch metadata again to that object
135+ schemas = self .get_schemas (object_name , session , metadata_url )
136+ if schemas [object_name ].get ("properties" ):
137+ input_args .update (** {"session" : session , "metadata_url" : metadata_url , "schemas" : schemas })
138+ return self .generate_stream (** input_args )
139+ retry_attempt += 1
140+ self .logger .warn (f"Object `{ object_name } )
141+ return None
115142
116143 def streams (self , config : Mapping [str , Any ]) -> List [Stream ]:
117144 auth = self .auth (config )
@@ -121,15 +148,15 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
121148 object_names = config .get ("object_types" )
122149
123150 # retrieve all record types if `object_types` config field is not specified
124- if not config . get ( "object_types" ) :
151+ if not object_names :
125152 objects_metadata = session .get (metadata_url ).json ().get ("items" )
126153 object_names = [object ["name" ] for object in objects_metadata ]
127154
128155 input_args = {"session" : session , "metadata_url" : metadata_url }
129156 schemas = self .get_schemas (object_names , ** input_args )
130157 input_args .update (
131158 ** {
132- "auth" : self . auth ( config ) ,
159+ "auth" : auth ,
133160 "base_url" : base_url ,
134161 "start_datetime" : config ["start_datetime" ],
135162 "window_in_days" : config ["window_in_days" ],
@@ -139,6 +166,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
139166 # build streams
140167 streams : list = []
141168 for name in object_names :
142- streams .append (self .generate_stream (object_name = name , ** input_args ))
143-
169+ stream = self .generate_stream (object_name = name .lower (), ** input_args )
170+ if stream :
171+ streams .append (stream )
144172 return streams
0 commit comments