1717 UpdateRecordData ,
1818 Upsert ,
1919)
20- from skyflow .generated .rest .core import ApiError
20+ from skyflow .generated .rest .core import ApiError , ParsingError
2121from skyflow .utils import SkyflowMessages , get_metrics
2222from skyflow .utils .enums import UpsertType
2323from skyflow .utils ._response_parsing import parse_tokens , parse_hashed_data , parse_metadata
@@ -127,7 +127,13 @@ def insert(self, request: InsertRequest, options: InsertOptions = None) -> Inser
127127 ]
128128 except Exception as e :
129129 log_error_log (SkyflowMessages .ErrorLogs .INSERT_RECORDS_REJECTED .value , self ._vault_client .get_logger ())
130- raise self .__to_skyflow_error (e )
130+ error_records , request_id = self .__unary_error_records (e )
131+ if error_records is None :
132+ raise self .__to_skyflow_error (e )
133+ records = [
134+ InsertResponseRecord (** self .__record_kwargs (record , include_data = False , request_id = request_id ))
135+ for record in error_records
136+ ]
131137
132138 log_info (SkyflowMessages .Info .INSERT_SUCCESS .value , self ._vault_client .get_logger ())
133139 return InsertResponse (records = records )
@@ -170,7 +176,13 @@ def get(self, request: GetRequest, options: GetOptions = None) -> GetResponse:
170176 ]
171177 except Exception as e :
172178 log_error_log (SkyflowMessages .ErrorLogs .GET_RECORDS_REJECTED .value , self ._vault_client .get_logger ())
173- raise self .__to_skyflow_error (e )
179+ error_records , request_id = self .__unary_error_records (e )
180+ if error_records is None :
181+ raise self .__to_skyflow_error (e )
182+ records = [
183+ GetResponseRecord (** self .__record_kwargs (record , include_data = True , request_id = request_id ))
184+ for record in error_records
185+ ]
174186
175187 log_info (SkyflowMessages .Info .GET_SUCCESS .value , self ._vault_client .get_logger ())
176188 return GetResponse (records = records )
@@ -212,7 +224,13 @@ def update(self, request: UpdateRequest, options: UpdateOptions = None) -> Updat
212224 ]
213225 except Exception as e :
214226 log_error_log (SkyflowMessages .ErrorLogs .UPDATE_RECORDS_REJECTED .value , self ._vault_client .get_logger ())
215- raise self .__to_skyflow_error (e )
227+ error_records , request_id = self .__unary_error_records (e )
228+ if error_records is None :
229+ raise self .__to_skyflow_error (e )
230+ records = [
231+ UpdateResponseRecord (** self .__record_kwargs (record , include_data = True , request_id = request_id ))
232+ for record in error_records
233+ ]
216234
217235 log_info (SkyflowMessages .Info .UPDATE_SUCCESS .value , self ._vault_client .get_logger ())
218236 return UpdateResponse (records = records )
@@ -240,7 +258,10 @@ def delete(self, request: DeleteRequest, options: DeleteOptions = None) -> Delet
240258 records = [self .__delete_row (record , request_id ) for record in (raw_response .data .records or [])]
241259 except Exception as e :
242260 log_error_log (SkyflowMessages .ErrorLogs .DELETE_RECORDS_REJECTED .value , self ._vault_client .get_logger ())
243- raise self .__to_skyflow_error (e )
261+ error_records , request_id = self .__unary_error_records (e )
262+ if error_records is None :
263+ raise self .__to_skyflow_error (e )
264+ records = [self .__delete_row (record , request_id ) for record in error_records ]
244265
245266 log_info (SkyflowMessages .Info .DELETE_SUCCESS .value , self ._vault_client .get_logger ())
246267 return DeleteResponse (records = records )
@@ -265,7 +286,10 @@ def detokenize(self, request: DetokenizeRequest, options: DetokenizeOptions = No
265286 records = [self .__detokenize_row (resp , request_id ) for resp in (raw_response .data .response or [])]
266287 except Exception as e :
267288 log_error_log (SkyflowMessages .ErrorLogs .DETOKENIZE_RECORDS_REJECTED .value , self ._vault_client .get_logger ())
268- raise self .__to_skyflow_error (e )
289+ error_records , request_id = self .__unary_error_records (e )
290+ if error_records is None :
291+ raise self .__to_skyflow_error (e )
292+ records = [self .__detokenize_row (resp , request_id ) for resp in error_records ]
269293
270294 log_info (SkyflowMessages .Info .DETOKENIZE_SUCCESS .value , self ._vault_client .get_logger ())
271295 return DetokenizeResponse (records = records )
@@ -450,15 +474,15 @@ def __format_bulk_insert_batch(self, records, start_index, headers):
450474 request_id = self .__extract_request_id (headers )
451475 rows = []
452476 for offset , record in enumerate (records ):
453- error = getattr (record , 'error' , None )
477+ error = self . __wire_record_value (record , 'error' , 'error' )
454478 rows .append (BulkInsertResponseRecord (
455479 index = start_index + offset ,
456480 request_id = request_id if error is not None else None ,
457- table_name = getattr (record , 'table_name ' , None ),
458- skyflow_id = getattr (record , 'skyflow_id ' , None ),
459- tokens = parse_tokens (getattr (record , 'tokens' , None )),
460- hashed_data = parse_hashed_data (getattr (record , 'hashed_data ' , None )),
461- http_code = getattr (record , 'http_code ' , None ),
481+ table_name = self . __wire_record_value (record , 'tableName ' , 'table_name' ),
482+ skyflow_id = self . __wire_record_value (record , 'skyflowID ' , 'skyflow_id' ),
483+ tokens = parse_tokens (self . __wire_record_value (record , 'tokens' , 'tokens' )),
484+ hashed_data = parse_hashed_data (self . __wire_record_value (record , 'hashedData ' , 'hashed_data' )),
485+ http_code = self . __wire_record_value (record , 'httpCode ' , 'http_code' ),
462486 error = error ,
463487 ))
464488 return rows
@@ -467,19 +491,44 @@ def __format_bulk_detokenize_batch(self, responses, start_index, headers):
467491 request_id = self .__extract_request_id (headers )
468492 rows = []
469493 for offset , resp in enumerate (responses ):
470- error = getattr (resp , 'error' , None )
494+ error = self . __wire_record_value (resp , 'error' , 'error' )
471495 rows .append (BulkDetokenizeResponseRecord (
472496 index = start_index + offset ,
473497 request_id = request_id if error is not None else None ,
474- value = getattr (resp , 'value' , None ),
475- token_group_name = getattr (resp , 'token_group_name ' , None ),
476- metadata = parse_metadata (getattr (resp , 'metadata' , None )),
477- http_code = getattr (resp , 'http_code ' , None ),
478- token = getattr (resp , 'token' , None ),
498+ value = self . __wire_record_value (resp , 'value' , 'value' ),
499+ token_group_name = self . __wire_record_value (resp , 'tokenGroupName ' , 'token_group_name' ),
500+ metadata = parse_metadata (self . __wire_record_value (resp , 'metadata' , 'metadata' )),
501+ http_code = self . __wire_record_value (resp , 'httpCode ' , 'http_code' ),
502+ token = self . __wire_record_value (resp , 'token' , 'token' ),
479503 error = error ,
480504 ))
481505 return rows
482506
507+ def __error_body_records (self , e ):
508+ body = getattr (e , 'body' , None )
509+ if isinstance (body , dict ):
510+ records = body .get ('records' )
511+ if records is None :
512+ records = body .get ('response' )
513+ else :
514+ records = getattr (body , 'records' , None ) or getattr (body , 'response' , None )
515+ if records is None :
516+ extra = getattr (body , '__pydantic_extra__' , None )
517+ if isinstance (extra , dict ):
518+ records = extra .get ('records' ) or extra .get ('response' )
519+ return records if isinstance (records , list ) and records else None
520+
521+ def __unary_error_records (self , e ):
522+ records = self .__error_body_records (e )
523+ if not records :
524+ return None , None
525+ return records , self .__extract_request_id (getattr (e , 'headers' , None ))
526+
527+ def __wire_record_value (self , record , wire_key , attr ):
528+ if isinstance (record , dict ):
529+ return record .get (wire_key )
530+ return getattr (record , attr , None )
531+
483532 def __bulk_batch_error_tuples (self , e , count , start_index ):
484533 if isinstance (e , ApiError ):
485534 request_id = self .__extract_request_id (e .headers )
@@ -500,13 +549,19 @@ def __bulk_batch_error_tuples(self, e, count, start_index):
500549 return [(start_index + i , None , message , None ) for i in range (count )]
501550
502551 def __bulk_insert_batch_error_rows (self , e , count , start_index ):
552+ records = self .__error_body_records (e )
553+ if records :
554+ return self .__format_bulk_insert_batch (records , start_index , getattr (e , 'headers' , None ))
503555 return [
504556 BulkInsertResponseRecord (index = idx , request_id = request_id , table_name = None , skyflow_id = None ,
505557 tokens = None , hashed_data = None , http_code = code , error = message )
506558 for idx , request_id , message , code in self .__bulk_batch_error_tuples (e , count , start_index )
507559 ]
508560
509561 def __bulk_detokenize_batch_error_rows (self , e , count , start_index ):
562+ records = self .__error_body_records (e )
563+ if records :
564+ return self .__format_bulk_detokenize_batch (records , start_index , getattr (e , 'headers' , None ))
510565 return [
511566 BulkDetokenizeResponseRecord (index = idx , request_id = request_id , value = None , token_group_name = None ,
512567 metadata = None , http_code = code , token = None , error = message )
@@ -620,24 +675,24 @@ def __extract_request_id(self, headers):
620675 return headers .get (REQUEST_ID_HEADER ) if headers else None
621676
622677 def __record_kwargs (self , record , include_data , request_id = None ):
623- error = getattr (record , 'error' , None )
678+ error = self . __wire_record_value (record , 'error' , 'error' )
624679 kwargs = {
625- 'table_name' : getattr (record , 'table_name ' , None ),
626- 'skyflow_id' : getattr (record , 'skyflow_id ' , None ),
627- 'tokens' : parse_tokens (getattr (record , 'tokens' , None )),
628- 'hashed_data' : parse_hashed_data (getattr (record , 'hashed_data ' , None )),
629- 'http_code' : getattr (record , 'http_code ' , None ),
680+ 'table_name' : self . __wire_record_value (record , 'tableName ' , 'table_name' ),
681+ 'skyflow_id' : self . __wire_record_value (record , 'skyflowID ' , 'skyflow_id' ),
682+ 'tokens' : parse_tokens (self . __wire_record_value (record , 'tokens' , 'tokens' )),
683+ 'hashed_data' : parse_hashed_data (self . __wire_record_value (record , 'hashedData ' , 'hashed_data' )),
684+ 'http_code' : self . __wire_record_value (record , 'httpCode ' , 'http_code' ),
630685 'error' : error ,
631686 'request_id' : request_id if error is not None else None ,
632687 }
633688 if include_data :
634- kwargs ['data' ] = getattr (record , 'data' , None )
689+ kwargs ['data' ] = self . __wire_record_value (record , 'data' , 'data' )
635690 return kwargs
636691
637692 def __to_skyflow_error (self , e ):
638693 if isinstance (e , SkyflowError ):
639694 return e
640- if isinstance (e , ApiError ):
695+ if isinstance (e , ( ApiError , ParsingError ) ):
641696 message , grpc_code , http_status , details = self .__parse_api_error_body (e .body )
642697 return SkyflowError (
643698 message = message ,
@@ -694,22 +749,22 @@ def __to_get_request_data(self, records):
694749 ]
695750
696751 def __delete_row (self , record , request_id = None ):
697- error = getattr (record , 'error' , None )
752+ error = self . __wire_record_value (record , 'error' , 'error' )
698753 return DeleteResponseRecord (
699- skyflow_id = getattr (record , 'skyflow_id ' , None ),
700- http_code = getattr (record , 'http_code ' , None ),
754+ skyflow_id = self . __wire_record_value (record , 'skyflowID ' , 'skyflow_id' ),
755+ http_code = self . __wire_record_value (record , 'httpCode ' , 'http_code' ),
701756 error = error ,
702757 request_id = request_id if error is not None else None ,
703758 )
704759
705760 def __detokenize_row (self , resp , request_id = None ):
706- error = getattr (resp , 'error' , None )
761+ error = self . __wire_record_value (resp , 'error' , 'error' )
707762 return DetokenizeResponseRecord (
708- token = getattr (resp , 'token' , None ),
709- value = getattr (resp , 'value' , None ),
710- token_group_name = getattr (resp , 'token_group_name ' , None ),
711- metadata = parse_metadata (getattr (resp , 'metadata' , None )),
712- http_code = getattr (resp , 'http_code ' , None ),
763+ token = self . __wire_record_value (resp , 'token' , 'token' ),
764+ value = self . __wire_record_value (resp , 'value' , 'value' ),
765+ token_group_name = self . __wire_record_value (resp , 'tokenGroupName ' , 'token_group_name' ),
766+ metadata = parse_metadata (self . __wire_record_value (resp , 'metadata' , 'metadata' )),
767+ http_code = self . __wire_record_value (resp , 'httpCode ' , 'http_code' ),
713768 error = error ,
714769 request_id = request_id if error is not None else None ,
715770 )
0 commit comments