Source code for langchain_community.tools.openapi.utils.api_models
"""Pydantic models for parsing an OpenAPI spec."""from__future__importannotationsimportloggingfromenumimportEnumfromtypingimport(TYPE_CHECKING,Any,Dict,List,Optional,Sequence,Tuple,Type,Union,)frompydanticimportBaseModel,Fieldfromlangchain_community.tools.openapi.utils.openapi_utilsimportHTTPVerb,OpenAPISpeclogger=logging.getLogger(__name__)PRIMITIVE_TYPES={"integer":int,"number":float,"string":str,"boolean":bool,"array":List,"object":Dict,"null":None,}# See https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#parameterIn# for more info.
[docs]classAPIPropertyLocation(Enum):"""The location of the property."""QUERY="query"PATH="path"HEADER="header"COOKIE="cookie"# Not yet supported@classmethoddeffrom_str(cls,location:str)->"APIPropertyLocation":"""Parse an APIPropertyLocation."""try:returncls(location)exceptValueError:raiseValueError(f"Invalid APIPropertyLocation. Valid values are {cls.__members__}")
_SUPPORTED_MEDIA_TYPES=("application/json",)SUPPORTED_LOCATIONS={APIPropertyLocation.HEADER,APIPropertyLocation.QUERY,APIPropertyLocation.PATH,}INVALID_LOCATION_TEMPL=('Unsupported APIPropertyLocation "{location}"'" for parameter {name}. "+f"Valid values are {[loc.valueforlocinSUPPORTED_LOCATIONS]}")SCHEMA_TYPE=Union[str,Type,tuple,None,Enum]
[docs]classAPIPropertyBase(BaseModel):"""Base model for an API property."""# The name of the parameter is required and is case-sensitive.# If "in" is "path", the "name" field must correspond to a template expression# within the path field in the Paths Object.# If "in" is "header" and the "name" field is "Accept", "Content-Type",# or "Authorization", the parameter definition is ignored.# For all other cases, the "name" corresponds to the parameter# name used by the "in" property.name:str=Field(alias="name")"""The name of the property."""required:bool=Field(alias="required")"""Whether the property is required."""type:SCHEMA_TYPE=Field(alias="type")"""The type of the property. Either a primitive type, a component/parameter type, or an array or 'object' (dict) of the above."""default:Optional[Any]=Field(alias="default",default=None)"""The default value of the property."""description:Optional[str]=Field(alias="description",default=None)"""The description of the property."""
[docs]classAPIProperty(APIPropertyBase):"""A model for a property in the query, path, header, or cookie params."""location:APIPropertyLocation=Field(alias="location")"""The path/how it's being passed to the endpoint."""@staticmethoddef_cast_schema_list_type(schema:Schema,)->Optional[Union[str,Tuple[str,...]]]:type_=schema.typeifnotisinstance(type_,list):returntype_else:returntuple(type_)@staticmethoddef_get_schema_type_for_enum(parameter:Parameter,schema:Schema)->Enum:"""Get the schema type when the parameter is an enum."""param_name=f"{parameter.name}Enum"returnEnum(param_name,{str(v):vforvinschema.enum})@staticmethoddef_get_schema_type_for_array(schema:Schema,)->Optional[Union[str,Tuple[str,...]]]:fromopenapi_pydanticimport(Reference,Schema,)items=schema.itemsifisinstance(items,Schema):schema_type=APIProperty._cast_schema_list_type(items)elifisinstance(items,Reference):ref_name=items.ref.split("/")[-1]schema_type=ref_name# TODO: Add ref definitions to make his validelse:raiseValueError(f"Unsupported array items: {items}")ifisinstance(schema_type,str):# TODO: recurseschema_type=(schema_type,)returnschema_type@staticmethoddef_get_schema_type(parameter:Parameter,schema:Optional[Schema])->SCHEMA_TYPE:ifschemaisNone:returnNoneschema_type:SCHEMA_TYPE=APIProperty._cast_schema_list_type(schema)ifschema_type=="array":schema_type=APIProperty._get_schema_type_for_array(schema)elifschema_type=="object":# TODO: Resolve array and object types to components.raiseNotImplementedError("Objects not yet supported")elifschema_typeinPRIMITIVE_TYPES:ifschema.enum:schema_type=APIProperty._get_schema_type_for_enum(parameter,schema)else:# Directly use the primitive typepasselse:raiseNotImplementedError(f"Unsupported type: {schema_type}")returnschema_type@staticmethoddef_validate_location(location:APIPropertyLocation,name:str)->None:iflocationnotinSUPPORTED_LOCATIONS:raiseNotImplementedError(INVALID_LOCATION_TEMPL.format(location=location,name=name))@staticmethoddef_validate_content(content:Optional[Dict[str,MediaType]])->None:ifcontent:raiseValueError("API Properties with media content not supported. ""Media content only supported within APIRequestBodyProperty's")@staticmethoddef_get_schema(parameter:Parameter,spec:OpenAPISpec)->Optional[Schema]:fromopenapi_pydanticimport(Reference,Schema,)schema=parameter.param_schemaifisinstance(schema,Reference):schema=spec.get_referenced_schema(schema)elifschemaisNone:returnNoneelifnotisinstance(schema,Schema):raiseValueError(f"Error dereferencing schema: {schema}")returnschema
[docs]@staticmethoddefis_supported_location(location:str)->bool:"""Return whether the provided location is supported."""try:returnAPIPropertyLocation.from_str(location)inSUPPORTED_LOCATIONSexceptValueError:returnFalse
[docs]@classmethoddeffrom_parameter(cls,parameter:Parameter,spec:OpenAPISpec)->"APIProperty":"""Instantiate from an OpenAPI Parameter."""location=APIPropertyLocation.from_str(parameter.param_in)cls._validate_location(location,parameter.name,)cls._validate_content(parameter.content)schema=cls._get_schema(parameter,spec)schema_type=cls._get_schema_type(parameter,schema)default_val=schema.defaultifschemaisnotNoneelseNonereturncls(name=parameter.name,location=location,default=default_val,description=parameter.description,required=parameter.required,type=schema_type,)
[docs]classAPIRequestBodyProperty(APIPropertyBase):"""A model for a request body property."""properties:List["APIRequestBodyProperty"]=Field(alias="properties")"""The sub-properties of the property."""# This is useful for handling nested property cycles.# We can define separate types in that case.references_used:List[str]=Field(alias="references_used")"""The references used by the property."""@classmethoddef_process_object_schema(cls,schema:Schema,spec:OpenAPISpec,references_used:List[str])->Tuple[Union[str,List[str],None],List["APIRequestBodyProperty"]]:fromopenapi_pydanticimport(Reference,)properties=[]required_props=schema.requiredor[]ifschema.propertiesisNone:raiseValueError(f"No properties found when processing object schema: {schema}")forprop_name,prop_schemainschema.properties.items():ifisinstance(prop_schema,Reference):ref_name=prop_schema.ref.split("/")[-1]ifref_namenotinreferences_used:references_used.append(ref_name)prop_schema=spec.get_referenced_schema(prop_schema)else:continueproperties.append(cls.from_schema(schema=prop_schema,name=prop_name,required=prop_nameinrequired_props,spec=spec,references_used=references_used,))returnschema.type,properties@classmethoddef_process_array_schema(cls,schema:Schema,name:str,spec:OpenAPISpec,references_used:List[str],)->str:fromopenapi_pydanticimportReference,Schemaitems=schema.itemsifitemsisnotNone:ifisinstance(items,Reference):ref_name=items.ref.split("/")[-1]ifref_namenotinreferences_used:references_used.append(ref_name)items=spec.get_referenced_schema(items)else:passreturnf"Array<{ref_name}>"else:passifisinstance(items,Schema):array_type=cls.from_schema(schema=items,name=f"{name}Item",required=True,# TODO: Add requiredspec=spec,references_used=references_used,)returnf"Array<{array_type.type}>"return"array"
[docs]@classmethoddeffrom_schema(cls,schema:Schema,name:str,required:bool,spec:OpenAPISpec,references_used:Optional[List[str]]=None,)->"APIRequestBodyProperty":"""Recursively populate from an OpenAPI Schema."""ifreferences_usedisNone:references_used=[]schema_type=schema.typeproperties:List[APIRequestBodyProperty]=[]ifschema_type=="object"andschema.properties:schema_type,properties=cls._process_object_schema(schema,spec,references_used)elifschema_type=="array":schema_type=cls._process_array_schema(schema,name,spec,references_used)elifschema_typeinPRIMITIVE_TYPES:# Use the primitive type directlypasselifschema_typeisNone:# No typing specified/parsed. WIll map to 'any'passelse:raiseValueError(f"Unsupported type: {schema_type}")returncls(name=name,required=required,type=schema_type,default=schema.default,description=schema.description,properties=properties,references_used=references_used,)
# class APIRequestBodyProperty(APIPropertyBase):
[docs]classAPIRequestBody(BaseModel):"""A model for a request body."""description:Optional[str]=Field(alias="description")"""The description of the request body."""properties:List[APIRequestBodyProperty]=Field(alias="properties")# E.g., application/json - we only support JSON at the moment.media_type:str=Field(alias="media_type")"""The media type of the request body."""@classmethoddef_process_supported_media_type(cls,media_type_obj:MediaType,spec:OpenAPISpec,)->List[APIRequestBodyProperty]:"""Process the media type of the request body."""fromopenapi_pydanticimportReferencereferences_used=[]schema=media_type_obj.media_type_schemaifisinstance(schema,Reference):references_used.append(schema.ref.split("/")[-1])schema=spec.get_referenced_schema(schema)ifschemaisNone:raiseValueError(f"Could not resolve schema for media type: {media_type_obj}")api_request_body_properties=[]required_properties=schema.requiredor[]ifschema.type=="object"andschema.properties:forprop_name,prop_schemainschema.properties.items():ifisinstance(prop_schema,Reference):prop_schema=spec.get_referenced_schema(prop_schema)api_request_body_properties.append(APIRequestBodyProperty.from_schema(schema=prop_schema,name=prop_name,required=prop_nameinrequired_properties,spec=spec,))else:api_request_body_properties.append(APIRequestBodyProperty(name="body",required=True,type=schema.type,default=schema.default,description=schema.description,properties=[],references_used=references_used,))returnapi_request_body_properties
[docs]@classmethoddeffrom_request_body(cls,request_body:RequestBody,spec:OpenAPISpec)->"APIRequestBody":"""Instantiate from an OpenAPI RequestBody."""properties=[]formedia_type,media_type_objinrequest_body.content.items():ifmedia_typenotin_SUPPORTED_MEDIA_TYPES:continueapi_request_body_properties=cls._process_supported_media_type(media_type_obj,spec,)properties.extend(api_request_body_properties)returncls(description=request_body.description,properties=properties,media_type=media_type,)
# class APIRequestBodyProperty(APIPropertyBase):# class APIRequestBody(BaseModel):
[docs]classAPIOperation(BaseModel):"""A model for a single API operation."""operation_id:str=Field(alias="operation_id")"""The unique identifier of the operation."""description:Optional[str]=Field(alias="description")"""The description of the operation."""base_url:str=Field(alias="base_url")"""The base URL of the operation."""path:str=Field(alias="path")"""The path of the operation."""method:HTTPVerb=Field(alias="method")"""The HTTP method of the operation."""properties:Sequence[APIProperty]=Field(alias="properties")# TODO: Add parse in used components to be able to specify what type of# referenced object it is.# """The properties of the operation."""# components: Dict[str, BaseModel] = Field(alias="components")request_body:Optional[APIRequestBody]=Field(alias="request_body")"""The request body of the operation."""@staticmethoddef_get_properties_from_parameters(parameters:List[Parameter],spec:OpenAPISpec)->List[APIProperty]:"""Get the properties of the operation."""properties=[]forparaminparameters:ifAPIProperty.is_supported_location(param.param_in):properties.append(APIProperty.from_parameter(param,spec))elifparam.required:raiseValueError(INVALID_LOCATION_TEMPL.format(location=param.param_in,name=param.name))else:logger.warning(INVALID_LOCATION_TEMPL.format(location=param.param_in,name=param.name)+" Ignoring optional parameter")passreturnproperties
[docs]@classmethoddeffrom_openapi_url(cls,spec_url:str,path:str,method:str,)->"APIOperation":"""Create an APIOperation from an OpenAPI URL."""spec=OpenAPISpec.from_url(spec_url)returncls.from_openapi_spec(spec,path,method)
[docs]@classmethoddeffrom_openapi_spec(cls,spec:OpenAPISpec,path:str,method:str,)->"APIOperation":"""Create an APIOperation from an OpenAPI spec."""operation=spec.get_operation(path,method)parameters=spec.get_parameters_for_operation(operation)properties=cls._get_properties_from_parameters(parameters,spec)operation_id=OpenAPISpec.get_cleaned_operation_id(operation,path,method)request_body=spec.get_request_body_for_operation(operation)api_request_body=(APIRequestBody.from_request_body(request_body,spec)ifrequest_bodyisnotNoneelseNone)description=operation.descriptionoroperation.summaryifnotdescriptionandspec.pathsisnotNone:description=spec.paths[path].descriptionorspec.paths[path].summaryreturncls(operation_id=operation_id,description=descriptionor"",base_url=spec.base_url,path=path,method=method,# type: ignore[arg-type]properties=properties,request_body=api_request_body,)
[docs]@staticmethoddefts_type_from_python(type_:SCHEMA_TYPE)->str:iftype_isNone:# TODO: Handle Nones better. These often result when# parsing specs that are < v3return"any"elifisinstance(type_,str):return{"str":"string","integer":"number","float":"number","date-time":"string",}.get(type_,type_)elifisinstance(type_,tuple):returnf"Array<{APIOperation.ts_type_from_python(type_[0])}>"elifisinstance(type_,type)andissubclass(type_,Enum):return" | ".join([f"'{e.value}'"foreintype_])else:returnstr(type_)