# SPDX-License-Identifier: Apache-2.0## The OpenSearch Contributors require contributions made to# this file be licensed under the Apache-2.0 license or a# compatible open source license.## Modifications Copyright OpenSearch Contributors. See# GitHub history for details.## Licensed to Elasticsearch B.V. under one or more contributor# license agreements. See the NOTICE file distributed with# this work for additional information regarding copyright# ownership. Elasticsearch B.V. licenses this file to you under# the Apache License, Version 2.0 (the "License"); you may# not use this file except in compliance with the License.# You may obtain a copy of the License at## http://www.apache.org/licenses/LICENSE-2.0## Unless required by applicable law or agreed to in writing,# software distributed under the License is distributed on an# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY# KIND, either express or implied. See the License for the# specific language governing permissions and limitations# under the License.fromtypingimportAny,Dict,Optionaltry:importsimplejsonasjsonexceptImportError:importjson# type: ignoreimportuuidfromdatetimeimportdate,datetimefromdecimalimportDecimalfrom.compatimportstring_typesfrom.exceptionsimportImproperlyConfigured,SerializationErrorfrom.helpers.utilsimportAttrListINTEGER_TYPES=()FLOAT_TYPES=(Decimal,)TIME_TYPES=(date,datetime)classSerializer:mimetype:str=""defloads(self,s:str)->Any:raiseNotImplementedError()defdumps(self,data:Any)->Any:raiseNotImplementedError()classTextSerializer(Serializer):mimetype:str="text/plain"defloads(self,s:str)->Any:returnsdefdumps(self,data:Any)->Any:ifisinstance(data,string_types):returndataraiseSerializationError(f"Cannot serialize {data!r} into text.")
[docs]classJSONSerializer(Serializer):mimetype:str="application/json"defdefault(self,data:Any)->Any:ifisinstance(data,TIME_TYPES):# Little hack to avoid importing pandas but to not# return 'NaT' string for pd.NaT as that's not a valid# date.formatted_data=data.isoformat()ifformatted_data!="NaT":returnformatted_dataifisinstance(data,uuid.UUID):returnstr(data)elifisinstance(data,FLOAT_TYPES):returnfloat(data)elifINTEGER_TYPESandisinstance(data,INTEGER_TYPES):returnint(data)# Special cases for numpy and pandas types# These are expensive to import so we try them last.try:importnumpyasnpifisinstance(data,(np.int_,np.intc,np.int8,np.int16,np.int32,np.int64,np.uint8,np.uint16,np.uint32,np.uint64,),):returnint(data)elifisinstance(data,(np.float16,np.float32,np.float64,),):returnfloat(data)elifisinstance(data,np.bool_):returnbool(data)elifisinstance(data,np.datetime64):returndata.item().isoformat()elifisinstance(data,np.ndarray):returndata.tolist()exceptImportError:passtry:importpandasaspdifisinstance(data,(pd.Series,pd.Categorical)):returndata.tolist()elifisinstance(data,pd.Timestamp)anddataisnotgetattr(pd,"NaT",None):returndata.isoformat()elifdataisgetattr(pd,"NA",None):returnNoneexceptImportError:passraiseTypeError(f"Unable to serialize {data!r} (type: {type(data)})")defloads(self,s:str)->Any:try:returnjson.loads(s)except(ValueError,TypeError)ase:raiseSerializationError(s,e)defdumps(self,data:Any)->Any:# don't serialize stringsifisinstance(data,string_types):returndatatry:returnjson.dumps(data,default=self.default,ensure_ascii=False,separators=(",",":"))except(ValueError,TypeError)ase:raiseSerializationError(data,e)
DEFAULT_SERIALIZERS:Dict[str,Serializer]={JSONSerializer.mimetype:JSONSerializer(),TextSerializer.mimetype:TextSerializer(),}classDeserializer:def__init__(self,serializers:Dict[str,Serializer],default_mimetype:str="application/json",)->None:try:self.default=serializers[default_mimetype]exceptKeyError:raiseImproperlyConfigured(f"Cannot find default serializer ({default_mimetype})")self.serializers=serializersdefloads(self,s:str,mimetype:Optional[str]=None)->Any:ifnotmimetype:deserializer=self.defaultelse:# Treat 'application/vnd.elasticsearch+json'# as application/json for compatibility.ifmimetype=="application/vnd.elasticsearch+json":mimetype="application/json"# split out charsetmimetype,_,_=mimetype.partition(";")try:deserializer=self.serializers[mimetype]exceptKeyError:raiseSerializationError(f"Unknown mimetype, unable to deserialize: {mimetype}")returndeserializer.loads(s)classAttrJSONSerializer(JSONSerializer):defdefault(self,data:Any)->Any:ifisinstance(data,AttrList):returndata._l_ifhasattr(data,"to_dict"):returndata.to_dict()returnsuper().default(data)serializer=AttrJSONSerializer()