diff --git a/generate/generate.py b/generate/generate.py index 4c8ddf9e2..04a7ca7c1 100755 --- a/generate/generate.py +++ b/generate/generate.py @@ -6,6 +6,7 @@ import re package_name = 'kittycad' + def main(): cwd = os.getcwd() path = os.path.join(cwd, 'spec.json') @@ -20,6 +21,433 @@ def main(): generateTypes(cwd, parser) # Generate the paths. + generatePaths(cwd, parser) + + +def generatePaths(cwd: str, parser: OpenApiParser): + # Make sure we have the directory. + path = os.path.join(cwd, 'kittycad', 'api') + os.makedirs(path, exist_ok=True) + + # Open the __init__.py file. + file_name = '__init__.py' + file_path = os.path.join(path, file_name) + f = open(file_path, 'w') + f.write("\"\"\" Contains methods for accessing the API \"\"\"\n") + # Close the file. + f.close() + + # Generate the directory/__init__.py for each of the tags. + tags = parser.data['tags'] + for tag in tags: + tag_name = tag['name'] + tag_description = tag['description'] + tag_path = os.path.join(path, tag_name) + # Esnure the directory exists. + os.makedirs(tag_path, exist_ok=True) + # Open the __init__.py file. + file_name = '__init__.py' + file_path = os.path.join(tag_path, file_name) + f = open(file_path, 'w') + f.write( + "\"\"\" Contains methods for accessing the " + + tag_name + + " API paths: " + + tag_description + + " \"\"\"\n") + # Close the file. + f.close() + + # Generate the paths. + data = parser.data + paths = data['paths'] + for p in paths: + for method in paths[p]: + endpoint = paths[p][method] + generatePath(path, p, method, endpoint) + + +def generatePath(path: str, name: str, method: str, endpoint: dict): + # Generate the path. + file_name = camel_to_snake(endpoint['operationId']) + '.py' + # Add the tag to the path if it exists. + if 'tags' in endpoint: + tag_name = endpoint['tags'][0] + path = os.path.join(path, tag_name) + file_path = os.path.join(path, file_name) + print("generating type: ", name, " at: ", file_path) + print(" endpoint: ", [endpoint]) + f = open(file_path, "w") + + endoint_refs = getEndpointRefs(endpoint) + parameter_refs = getParameterRefs(endpoint) + request_body_refs = getRequestBodyRefs(endpoint) + + # Add our imports. + f.write("from typing import Any, Dict, Optional, Union\n") + f.write("\n") + f.write("import httpx\n") + f.write("\n") + f.write("from ...client import Client\n") + # Import our references for responses. + for ref in endoint_refs: + f.write( + "from ...models." + + camel_to_snake(ref) + + " import " + + ref + + "\n") + for ref in parameter_refs: + f.write( + "from ...models." + + camel_to_snake(ref) + + " import " + + ref + + "\n") + for ref in request_body_refs: + f.write( + "from ...models." + + camel_to_snake(ref) + + " import " + + ref + + "\n") + f.write("from ...types import Response\n") + f.write("\n") + + # Define the method. + f.write("def _get_kwargs(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + ": " + + parameter_type + + ",\n") + f.write("*, client: Client) -> Dict[str, Any]:\n") + f.write("\turl = \"{}" + name + "\".format(client.base_url,\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + parameter_name + + "=" + + camel_to_snake(parameter_name) + + ",\n") + f.write("\t)\n") + f.write("\n") + f.write("\theaders: Dict[str, Any] = client.get_headers()\n") + f.write("\tcookies: Dict[str, Any] = client.get_cookies()\n") + f.write("\n") + + f.write("\treturn {\n") + f.write("\t\t\"url\": url,\n") + f.write("\t\t\"headers\": headers,\n") + f.write("\t\t\"cookies\": cookies,\n") + f.write("\t\t\"timeout\": client.get_timeout(),\n") + f.write("\t}\n") + + # Define the parse reponse. + f.write("\n") + f.write("\n") + + f.write( + "def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + # Iterate over the responses. + responses = endpoint['responses'] + for response_code in responses: + response = responses[response_code] + f.write("\tif response.status_code == " + response_code + ":\n") + if 'content' in response: + content = response['content'] + for content_type in content: + if content_type == 'application/json': + json = content[content_type]['schema'] + if '$ref' in json: + ref = json['$ref'].replace('#/components/schemas/', '') + f.write( + "\t\tresponse_" + + response_code + + " = " + + ref + + ".from_dict(response.json())\n") + else: + f.write("\t\tresponse_" + response_code + " = None\n") + + f.write("\t\treturn response_" + response_code + "\n") + + # End the method. + f.write("\treturn None\n") + + # Define the build response method. + f.write("\n") + f.write("\n") + f.write( + "def _build_response(*, response: httpx.Response) -> Response[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + f.write("\treturn Response(\n") + f.write("\t\tstatus_code=response.status_code,\n") + f.write("\t\tcontent=response.content,\n") + f.write("\t\theaders=response.headers,\n") + f.write("\t\tparsed=_parse_response(response=response),\n") + f.write("\t)\n") + + # Define the sync_detailed method. + f.write("\n") + f.write("\n") + f.write( + "def sync_detailed(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + ": " + + parameter_type + + ",\n") + f.write("*, client: Client) -> Response[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + f.write("\tkwargs = _get_kwargs(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + "=" + + camel_to_snake(parameter_name) + + ",\n") + f.write("\t\tclient=client,\n") + f.write("\t)\n") + f.write("\n") + f.write("\tresponse = httpx." + method + "(\n") + f.write("\t\tverify=client.verify_ssl,\n") + f.write("\t\t**kwargs,\n") + f.write("\t)\n") + f.write("\n") + f.write("\treturn _build_response(response=response)\n") + + # Define the sync method. + f.write("\n") + f.write("\n") + f.write( + "def sync(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + ": " + + parameter_type + + ",\n") + f.write("*, client: Client) -> Optional[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + if 'description' in endpoint: + f.write("\t\"\"\" " + endpoint['description'] + " \"\"\"\n") + f.write("\n") + f.write("\treturn sync_detailed(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + "=" + + camel_to_snake(parameter_name) + + ",\n") + f.write("\t\tclient=client,\n") + f.write("\t).parsed\n") + + # Define the asyncio_detailed method. + f.write("\n") + f.write("\n") + f.write( + "async def asyncio_detailed(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + ": " + + parameter_type + + ",\n") + f.write("*, client: Client) -> Response[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + f.write("\tkwargs = _get_kwargs(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + "=" + + camel_to_snake(parameter_name) + + ",\n") + f.write("\t\tclient=client,\n") + f.write("\t)\n") + f.write("\n") + f.write("\tasync with httpx.AsyncClient(verify=client.verify_ssl) as _client:\n") + f.write("\t\tresponse = await _client." + method + "(**kwargs)\n") + f.write("\n") + f.write("\treturn _build_response(response=response)\n") + + # Define the asyncio method. + f.write("\n") + f.write("\n") + f.write( + "async def asyncio(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + ": " + + parameter_type + + ",\n") + f.write("*, client: Client) -> Optional[Union[Any, " + + ", ".join(endoint_refs) + + "]]:\n") + if 'description' in endpoint: + f.write("\t\"\"\" " + endpoint['description'] + " \"\"\"\n") + f.write("\n") + f.write("\treturn (await asyncio_detailed(\n") + # Iterate over the parameters. + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if 'type' in parameter['schema']: + parameter_type = parameter['schema']['type'].replace( + 'string', 'str') + elif '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + else: + print(" parameter: ", parameter) + raise Exception("Unknown parameter type") + f.write( + "\t" + + camel_to_snake(parameter_name) + + "=" + + camel_to_snake(parameter_name) + + ",\n") + f.write("\t\tclient=client,\n") + f.write("\t)).parsed\n") + + # Close the file. + f.close() + def generateTypes(cwd: str, parser: OpenApiParser): # Make sure we have the directory. @@ -39,11 +467,12 @@ def generateTypes(cwd: str, parser: OpenApiParser): for key in schemas: schema = schemas[key] generateType(path, key, schema) - f.write("from ."+camel_to_snake(key)+" import " + key + "\n") + f.write("from ." + camel_to_snake(key) + " import " + key + "\n") # Close the file. f.close() + def generateType(path: str, name: str, schema: dict): # Generate the type. file_name = camel_to_snake(name) + '.py' @@ -65,14 +494,19 @@ def generateType(path: str, name: str, schema: dict): refs = getRefs(schema) for ref in refs: - f.write("from ..models."+camel_to_snake(ref)+" import "+ref+"\n") + f.write( + "from ..models." + + camel_to_snake(ref) + + " import " + + ref + + "\n") f.write("from ..types import UNSET, Unset\n") f.write("\n") - f.write("T = TypeVar(\"T\", bound=\""+name+"\")\n") + f.write("T = TypeVar(\"T\", bound=\"" + name + "\")\n") f.write("\n") f.write("@attr.s(auto_attribs=True)\n") - f.write("class "+name+":\n") + f.write("class " + name + ":\n") # Write the description. f.write("\t\"\"\" \"\"\"\n") # Iterate over the properties. @@ -85,27 +519,49 @@ def generateType(path: str, name: str, schema: dict): if property_type == 'string': if 'format' in property_schema: if property_schema['format'] == 'date-time': - f.write("\t"+property_name+": Union[Unset, datetime.datetime] = UNSET\n") + f.write( + "\t" + + property_name + + ": Union[Unset, datetime.datetime] = UNSET\n") continue - f.write("\t"+property_name+": Union[Unset, str] = UNSET\n") + f.write( + "\t" + + property_name + + ": Union[Unset, str] = UNSET\n") elif property_type == 'integer': - f.write("\t"+property_name+": Union[Unset, int] = UNSET\n") + f.write( + "\t" + + property_name + + ": Union[Unset, int] = UNSET\n") elif property_type == 'number': - f.write("\t"+property_name+": Union[Unset, float] = UNSET\n") + f.write( + "\t" + + property_name + + ": Union[Unset, float] = UNSET\n") elif property_type == 'boolean': - f.write("\t"+property_name+": Union[Unset, bool] = False\n") + f.write( + "\t" + + property_name + + ": Union[Unset, bool] = False\n") else: raise (" unknown type: ", property_type) elif '$ref' in property_schema: - ref = property_schema['$ref'].replace('#/components/schemas/', '') - f.write("\t"+property_name+": Union[Unset, "+ref+"] = UNSET\n") + ref = property_schema['$ref'].replace( + '#/components/schemas/', '') + f.write( + "\t" + + property_name + + ": Union[Unset, " + + ref + + "] = UNSET\n") else: raise (" unknown schema: ", property_schema) # Finish writing the class. f.write("\n") - f.write("\tadditional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)\n") + f.write( + "\tadditional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)\n") # Now let's write the to_dict method. f.write("\n") @@ -120,25 +576,66 @@ def generateType(path: str, name: str, schema: dict): if property_type == 'string': if 'format' in property_schema: if property_schema['format'] == 'date-time': - f.write("\t\t"+property_name+": Union[Unset, str] = UNSET\n") - f.write("\t\tif not isinstance(self."+property_name+", Unset):\n") - f.write("\t\t\t"+property_name+" = self."+property_name+".isoformat()\n") + f.write( + "\t\t" + + property_name + + ": Union[Unset, str] = UNSET\n") + f.write( + "\t\tif not isinstance(self." + property_name + ", Unset):\n") + f.write( + "\t\t\t" + + property_name + + " = self." + + property_name + + ".isoformat()\n") continue - f.write("\t"+property_name+" = self."+property_name+"\n") + f.write( + "\t" + + property_name + + " = self." + + property_name + + "\n") elif property_type == 'integer': - f.write("\t"+property_name+" = self."+property_name+"\n") + f.write( + "\t" + + property_name + + " = self." + + property_name + + "\n") elif property_type == 'number': - f.write("\t"+property_name+" = self."+property_name+"\n") + f.write( + "\t" + + property_name + + " = self." + + property_name + + "\n") elif property_type == 'boolean': - f.write("\t"+property_name+" = self."+property_name+"\n") + f.write( + "\t" + + property_name + + " = self." + + property_name + + "\n") else: raise (" unknown type: ", property_type) elif '$ref' in property_schema: - ref = property_schema['$ref'].replace('#/components/schemas/', '') - f.write("\t\t"+property_name+": Union[Unset, str] = UNSET\n") - f.write("\t\tif not isinstance(self."+property_name+", Unset):\n") - f.write("\t\t\t"+property_name+" = self."+property_name+".value\n") + ref = property_schema['$ref'].replace( + '#/components/schemas/', '') + f.write( + "\t\t" + + property_name + + ": Union[Unset, str] = UNSET\n") + f.write( + "\t\tif not isinstance(self." + + property_name + + ", Unset):\n") + f.write( + "\t\t\t" + + property_name + + " = self." + + property_name + + ".value\n") else: raise (" unknown schema: ", property_schema) @@ -148,12 +645,16 @@ def generateType(path: str, name: str, schema: dict): f.write("\t\tfield_dict.update(self.additional_properties)\n") f.write("\t\tfield_dict.update({})\n") - # Iternate over the properties. for property_name in schema['properties']: # Write the property. - f.write("\t\tif "+property_name+" is not UNSET:\n") - f.write("\t\t\tfield_dict['"+property_name+"'] = "+property_name+"\n") + f.write("\t\tif " + property_name + " is not UNSET:\n") + f.write( + "\t\t\tfield_dict['" + + property_name + + "'] = " + + property_name + + "\n") f.write("\n") f.write("\t\treturn field_dict\n") @@ -161,7 +662,8 @@ def generateType(path: str, name: str, schema: dict): # Now let's write the from_dict method. f.write("\n") f.write("\t@classmethod\n") - f.write("\tdef from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:\n") + f.write( + "\tdef from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:\n") f.write("\t\td = src_dict.copy()\n") # Iternate over the properties. @@ -174,37 +676,78 @@ def generateType(path: str, name: str, schema: dict): if property_type == 'string': if 'format' in property_schema: if property_schema['format'] == 'date-time': - f.write("\t\t_"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") - f.write("\t\t"+property_name+": Union[Unset, datetime.datetime]\n") - f.write("\t\tif not isinstance(_"+property_name+", Unset):\n") - f.write("\t\t\t"+property_name+" = UNSET\n") + f.write( + "\t\t_" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") + f.write( + "\t\t" + + property_name + + ": Union[Unset, datetime.datetime]\n") + f.write( + "\t\tif not isinstance(_" + property_name + ", Unset):\n") + f.write("\t\t\t" + property_name + " = UNSET\n") f.write("\t\telse:\n") - f.write("\t\t\t"+property_name+" = isoparse(_"+property_name+")\n") + f.write("\t\t\t" + property_name + + " = isoparse(_" + property_name + ")\n") f.write("\n") continue - f.write("\t"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") + f.write( + "\t" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") f.write("\n") elif property_type == 'integer': - f.write("\t"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") + f.write( + "\t" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") f.write("\n") elif property_type == 'number': - f.write("\t"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") + f.write( + "\t" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") f.write("\n") elif property_type == 'boolean': - f.write("\t"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") + f.write( + "\t" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") f.write("\n") else: print(" unknown type: ", property_type) raise elif '$ref' in property_schema: - ref = property_schema['$ref'].replace('#/components/schemas/', '') - f.write("\t\t_"+property_name+" = d.pop(\"" +property_name+"\", UNSET)\n") - f.write("\t\t"+property_name+": Union[Unset, "+ref+"]\n") - f.write("\t\tif not isinstance(_"+property_name+", Unset):\n") - f.write("\t\t\t"+property_name+" = UNSET\n") + ref = property_schema['$ref'].replace( + '#/components/schemas/', '') + f.write( + "\t\t_" + + property_name + + " = d.pop(\"" + + property_name + + "\", UNSET)\n") + f.write("\t\t" + property_name + + ": Union[Unset, " + ref + "]\n") + f.write( + "\t\tif not isinstance(_" + + property_name + + ", Unset):\n") + f.write("\t\t\t" + property_name + " = UNSET\n") f.write("\t\telse:\n") - f.write("\t\t\t"+property_name+" = "+ref+"(_"+property_name+")\n") + f.write("\t\t\t" + property_name + " = " + + ref + "(_" + property_name + ")\n") f.write("\n") else: print(" unknown schema: ", property_schema) @@ -212,17 +755,17 @@ def generateType(path: str, name: str, schema: dict): # Finish writing the from_dict method. f.write("\n") - f.write("\t\t"+camel_to_snake(name)+" = cls(\n") + f.write("\t\t" + camel_to_snake(name) + " = cls(\n") # Iternate over the properties. for property_name in schema['properties']: # Write the property. - f.write("\t\t\t"+property_name+"= "+property_name+",\n") + f.write("\t\t\t" + property_name + "= " + property_name + ",\n") # Close the class. f.write("\t\t)\n") f.write("\n") - f.write("\t\t"+camel_to_snake(name)+".additional_properties = d\n") - f.write("return "+camel_to_snake(name)+"\n") + f.write("\t\t" + camel_to_snake(name) + ".additional_properties = d\n") + f.write("return " + camel_to_snake(name) + "\n") # write the rest of the class. f.write("\n") @@ -248,10 +791,15 @@ def generateType(path: str, name: str, schema: dict): elif type_name == 'string' and 'enum' in schema: f.write("from enum import Enum\n") f.write("\n") - f.write("class "+name+"(str, Enum):\n") + f.write("class " + name + "(str, Enum):\n") # Iterate over the properties. for value in schema['enum']: - f.write("\t"+camel_to_screaming_snake(value)+" = '"+value+"'\n") + f.write( + "\t" + + camel_to_screaming_snake(value) + + " = '" + + value + + "'\n") # close the enum. f.write("\n") @@ -264,6 +812,7 @@ def generateType(path: str, name: str, schema: dict): # Close the file. f.close() + def hasDateTime(schema: dict) -> bool: # Generate the type. if 'type' in schema: @@ -281,6 +830,7 @@ def hasDateTime(schema: dict) -> bool: return False + def getRefs(schema: dict) -> [str]: refs = [] if '$ref' in schema: @@ -301,14 +851,66 @@ def getRefs(schema: dict) -> [str]: return refs +def getEndpointRefs(endpoint: dict) -> [str]: + refs = [] + + responses = endpoint['responses'] + for response_code in responses: + response = responses[response_code] + if 'content' in response: + content = response['content'] + for content_type in content: + if content_type == 'application/json': + json = content[content_type]['schema'] + if '$ref' in json: + ref = json['$ref'].replace('#/components/schemas/', '') + refs.append(ref) + + return refs + + +def getParameterRefs(endpoint: dict) -> [str]: + refs = [] + + if 'parameters' in endpoint: + parameters = endpoint['parameters'] + for parameter in parameters: + parameter_name = parameter['name'] + if '$ref' in parameter['schema']: + parameter_type = parameter['schema']['$ref'].replace( + '#/components/schemas/', '') + refs.append(parameter_type) + + return refs + + +def getRequestBodyRefs(endpoint: dict) -> [str]: + refs = [] + + if 'requestBody' in endpoint: + requestBody = endpoint['requestBody'] + if 'content' in requestBody: + content = requestBody['content'] + for content_type in content: + if content_type == 'application/json': + json = content[content_type]['schema'] + if '$ref' in json: + ref = json['$ref'].replace('#/components/schemas/', '') + refs.append(ref) + + return refs + + def camel_to_snake(name: str): name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) return re.sub('([a-z0-9])([A-Z])', r'\1_\2', name).lower() + def camel_to_screaming_snake(name: str): name = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', name) return re.sub('([a-z0-9])([A-Z])', r'\1_\2', name).upper() + if (__name__ == '__main__'): exit_code = main() exit(exit_code) diff --git a/kittycad/api/beta/__init__.py b/kittycad/api/beta/__init__.py new file mode 100644 index 000000000..f80dbd97b --- /dev/null +++ b/kittycad/api/beta/__init__.py @@ -0,0 +1 @@ +""" Contains methods for accessing the beta API paths: Beta API endpoints. """ diff --git a/kittycad/api/file/__init__.py b/kittycad/api/file/__init__.py index e69de29bb..bd83ff7bb 100644 --- a/kittycad/api/file/__init__.py +++ b/kittycad/api/file/__init__.py @@ -0,0 +1 @@ +""" Contains methods for accessing the file API paths: CAD file operations. """ diff --git a/kittycad/api/file/file_conversion_by_id.py b/kittycad/api/file/file_conversion_by_id.py deleted file mode 100644 index b98ebd0f7..000000000 --- a/kittycad/api/file/file_conversion_by_id.py +++ /dev/null @@ -1,128 +0,0 @@ -from typing import Any, Dict, Optional, Union - -import httpx - -from ...client import AuthenticatedClient -from ...models.file_conversion import FileConversion -from ...types import Response - - -def _get_kwargs( - id: str, - *, - client: AuthenticatedClient, -) -> Dict[str, Any]: - url = "{}/file/conversion/{id}".format(client.base_url, id=id) - - headers: Dict[str, Any] = client.get_headers() - cookies: Dict[str, Any] = client.get_cookies() - - return { - "url": url, - "headers": headers, - "cookies": cookies, - "timeout": client.get_timeout(), - } - - -def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion]]: - if response.status_code == 200: - response_200 = FileConversion.from_dict(response.json()) - - return response_200 - if response.status_code == 400: - response_400 = None - - return response_400 - if response.status_code == 401: - response_401 = None - - return response_401 - if response.status_code == 403: - response_403 = None - - return response_403 - if response.status_code == 404: - response_404 = None - - return response_404 - if response.status_code == 406: - response_406 = None - - return response_406 - if response.status_code == 500: - response_500 = None - - return response_500 - return None - - -def _build_response(*, response: httpx.Response) -> Response[Union[Any, FileConversion]]: - return Response( - status_code=response.status_code, - content=response.content, - headers=response.headers, - parsed=_parse_response(response=response), - ) - - -def sync_detailed( - id: str, - *, - client: AuthenticatedClient, -) -> Response[Union[Any, FileConversion]]: - kwargs = _get_kwargs( - id=id, - client=client, - ) - - response = httpx.get( - verify=client.verify_ssl, - **kwargs, - ) - - return _build_response(response=response) - - -def sync( - id: str, - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, FileConversion]]: - """Get the status of a file conversion.""" - - return sync_detailed( - id=id, - client=client, - ).parsed - - -async def asyncio_detailed( - id: str, - *, - client: AuthenticatedClient, -) -> Response[Union[Any, FileConversion]]: - kwargs = _get_kwargs( - id=id, - client=client, - ) - - async with httpx.AsyncClient(verify=client.verify_ssl) as _client: - response = await _client.get(**kwargs) - - return _build_response(response=response) - - -async def asyncio( - id: str, - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, FileConversion]]: - """Get the status of a file conversion.""" - - return ( - await asyncio_detailed( - id=id, - client=client, - ) - ).parsed diff --git a/kittycad/api/file/file_conversion_status.py b/kittycad/api/file/file_conversion_status.py new file mode 100644 index 000000000..b67605a86 --- /dev/null +++ b/kittycad/api/file/file_conversion_status.py @@ -0,0 +1,107 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...models.file_conversion import FileConversion +from ...types import Response + +def _get_kwargs( + id: str, +*, client: Client) -> Dict[str, Any]: + url = "{}/file/conversion/{id}".format(client.base_url, + id=id, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion]]: + if response.status_code == 200: + response_200 = FileConversion.from_dict(response.json()) + return response_200 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + if response.status_code == 404: + response_404 = None + return response_404 + if response.status_code == 406: + response_406 = None + return response_406 + if response.status_code == 500: + response_500 = None + return response_500 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, FileConversion]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( + id: str, +*, client: Client) -> Response[Union[Any, FileConversion]]: + kwargs = _get_kwargs( + id=id, + client=client, + ) + + response = httpx.get( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( + id: str, +*, client: Client) -> Optional[Union[Any, FileConversion]]: + """ Get the status and output of an async file conversion. """ + + return sync_detailed( + id=id, + client=client, + ).parsed + + +async def asyncio_detailed( + id: str, +*, client: Client) -> Response[Union[Any, FileConversion]]: + kwargs = _get_kwargs( + id=id, + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.get(**kwargs) + + return _build_response(response=response) + + +async def asyncio( + id: str, +*, client: Client) -> Optional[Union[Any, FileConversion]]: + """ Get the status and output of an async file conversion. """ + + return (await asyncio_detailed( + id=id, + client=client, + )).parsed diff --git a/kittycad/api/file/file_convert.py b/kittycad/api/file/file_convert.py deleted file mode 100644 index 900cbb5b3..000000000 --- a/kittycad/api/file/file_convert.py +++ /dev/null @@ -1,150 +0,0 @@ -from typing import Any, Dict, Optional, Union - -import httpx - -from ...client import AuthenticatedClient -from ...models.file_conversion import FileConversion -from ...models.valid_file_type import ValidFileType -from ...types import Response - - -def _get_kwargs( - source_format: ValidFileType, - output_format: ValidFileType, - content: bytes, - *, - client: AuthenticatedClient, -) -> Dict[str, Any]: - url = "{}/file/conversion/{sourceFormat}/{outputFormat}".format( - client.base_url, sourceFormat=source_format, outputFormat=output_format - ) - - headers: Dict[str, Any] = client.get_headers() - cookies: Dict[str, Any] = client.get_cookies() - - return { - "url": url, - "headers": headers, - "cookies": cookies, - "timeout": client.get_timeout(), - "content": content, - } - - -def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion]]: - if response.status_code == 200: - response_200 = FileConversion.from_dict(response.json()) - - return response_200 - if response.status_code == 202: - response_202 = FileConversion.from_dict(response.json()) - - return response_202 - if response.status_code == 400: - response_400 = None - - return response_400 - if response.status_code == 401: - response_401 = None - - return response_401 - if response.status_code == 403: - response_403 = None - - return response_403 - if response.status_code == 406: - response_406 = None - - return response_406 - if response.status_code == 500: - response_500 = None - - return response_500 - return None - - -def _build_response(*, response: httpx.Response) -> Response[Union[Any, FileConversion]]: - return Response( - status_code=response.status_code, - content=response.content, - headers=response.headers, - parsed=_parse_response(response=response), - ) - - -def sync_detailed( - source_format: ValidFileType, - output_format: ValidFileType, - content: bytes, - *, - client: AuthenticatedClient, -) -> Response[Union[Any, FileConversion]]: - kwargs = _get_kwargs( - source_format=source_format, - output_format=output_format, - content=content, - client=client, - ) - - response = httpx.post( - verify=client.verify_ssl, - **kwargs, - ) - - return _build_response(response=response) - - -def sync( - source_format: ValidFileType, - output_format: ValidFileType, - content: bytes, - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, FileConversion]]: - """Convert a CAD file from one format to another. If the file being converted is larger than a certain size it will be performed asynchronously.""" - - return sync_detailed( - source_format=source_format, - output_format=output_format, - content=content, - client=client, - ).parsed - - -async def asyncio_detailed( - source_format: ValidFileType, - output_format: ValidFileType, - content: bytes, - *, - client: AuthenticatedClient, -) -> Response[Union[Any, FileConversion]]: - kwargs = _get_kwargs( - source_format=source_format, - output_format=output_format, - content=content, - client=client, - ) - - async with httpx.AsyncClient(verify=client.verify_ssl) as _client: - response = await _client.post(**kwargs) - - return _build_response(response=response) - - -async def asyncio( - source_format: ValidFileType, - output_format: ValidFileType, - content: bytes, - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, FileConversion]]: - """Convert a CAD file from one format to another. If the file being converted is larger than a certain size it will be performed asynchronously.""" - - return ( - await asyncio_detailed( - source_format=source_format, - output_format=output_format, - content=content, - client=client, - ) - ).parsed diff --git a/kittycad/api/file/post_file_conversion.py b/kittycad/api/file/post_file_conversion.py new file mode 100644 index 000000000..cc0d07373 --- /dev/null +++ b/kittycad/api/file/post_file_conversion.py @@ -0,0 +1,123 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...models.file_conversion import FileConversion +from ...models.file_conversion import FileConversion +from ...models.valid_source_file_format import ValidSourceFileFormat +from ...models.valid_output_file_format import ValidOutputFileFormat +from ...types import Response + +def _get_kwargs( + source_format: ValidSourceFileFormat, + output_format: ValidOutputFileFormat, +*, client: Client) -> Dict[str, Any]: + url = "{}/file/conversion/{sourceFormat}/{outputFormat}".format(client.base_url, + sourceFormat=source_format, + outputFormat=output_format, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion, FileConversion]]: + if response.status_code == 200: + response_200 = FileConversion.from_dict(response.json()) + return response_200 + if response.status_code == 202: + response_202 = FileConversion.from_dict(response.json()) + return response_202 + if response.status_code == 400: + response_400 = None + return response_400 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + if response.status_code == 406: + response_406 = None + return response_406 + if response.status_code == 500: + response_500 = None + return response_500 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, FileConversion, FileConversion]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( + source_format: ValidSourceFileFormat, + output_format: ValidOutputFileFormat, +*, client: Client) -> Response[Union[Any, FileConversion, FileConversion]]: + kwargs = _get_kwargs( + source_format=source_format, + output_format=output_format, + client=client, + ) + + response = httpx.post( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( + source_format: ValidSourceFileFormat, + output_format: ValidOutputFileFormat, +*, client: Client) -> Optional[Union[Any, FileConversion, FileConversion]]: + """ Convert a CAD file from one format to another. If the file being converted is larger than 30MB, it will be performed asynchronously. """ + + return sync_detailed( + source_format=source_format, + output_format=output_format, + client=client, + ).parsed + + +async def asyncio_detailed( + source_format: ValidSourceFileFormat, + output_format: ValidOutputFileFormat, +*, client: Client) -> Response[Union[Any, FileConversion, FileConversion]]: + kwargs = _get_kwargs( + source_format=source_format, + output_format=output_format, + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.post(**kwargs) + + return _build_response(response=response) + + +async def asyncio( + source_format: ValidSourceFileFormat, + output_format: ValidOutputFileFormat, +*, client: Client) -> Optional[Union[Any, FileConversion, FileConversion]]: + """ Convert a CAD file from one format to another. If the file being converted is larger than 30MB, it will be performed asynchronously. """ + + return (await asyncio_detailed( + source_format=source_format, + output_format=output_format, + client=client, + )).parsed diff --git a/kittycad/api/internal/__init__.py b/kittycad/api/internal/__init__.py new file mode 100644 index 000000000..3ecabd4f5 --- /dev/null +++ b/kittycad/api/internal/__init__.py @@ -0,0 +1 @@ +""" Contains methods for accessing the internal API paths: Internal API endpoints. """ diff --git a/kittycad/api/internal/gpu_devices.py b/kittycad/api/internal/gpu_devices.py new file mode 100644 index 000000000..f8bb3ca8d --- /dev/null +++ b/kittycad/api/internal/gpu_devices.py @@ -0,0 +1,86 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...types import Response + +def _get_kwargs( +*, client: Client) -> Dict[str, Any]: + url = "{}/_internal/gpu/devices".format(client.base_url, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, ]]: + if response.status_code == 200: + return response_200 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, ]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( +*, client: Client) -> Response[Union[Any, ]]: + kwargs = _get_kwargs( + client=client, + ) + + response = httpx.get( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( +*, client: Client) -> Optional[Union[Any, ]]: + """ Get information about GPU devices on this server. This is primarily used for debugging. This endpoint can only be used by specific KittyCAD employees. """ + + return sync_detailed( + client=client, + ).parsed + + +async def asyncio_detailed( +*, client: Client) -> Response[Union[Any, ]]: + kwargs = _get_kwargs( + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.get(**kwargs) + + return _build_response(response=response) + + +async def asyncio( +*, client: Client) -> Optional[Union[Any, ]]: + """ Get information about GPU devices on this server. This is primarily used for debugging. This endpoint can only be used by specific KittyCAD employees. """ + + return (await asyncio_detailed( + client=client, + )).parsed diff --git a/kittycad/api/internal/stop_async_conversions.py b/kittycad/api/internal/stop_async_conversions.py new file mode 100644 index 000000000..053639d47 --- /dev/null +++ b/kittycad/api/internal/stop_async_conversions.py @@ -0,0 +1,91 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...models.file_conversion import FileConversion +from ...types import Response + +def _get_kwargs( +*, client: Client) -> Dict[str, Any]: + url = "{}/_internal/async/conversions/stop".format(client.base_url, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion]]: + if response.status_code == 200: + response_200 = FileConversion.from_dict(response.json()) + return response_200 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + if response.status_code == 404: + response_404 = None + return response_404 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, FileConversion]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( +*, client: Client) -> Response[Union[Any, FileConversion]]: + kwargs = _get_kwargs( + client=client, + ) + + response = httpx.post( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( +*, client: Client) -> Optional[Union[Any, FileConversion]]: + """ Stop all async conversions that are currently running. This endpoint can only be used by specific KittyCAD employees. """ + + return sync_detailed( + client=client, + ).parsed + + +async def asyncio_detailed( +*, client: Client) -> Response[Union[Any, FileConversion]]: + kwargs = _get_kwargs( + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.post(**kwargs) + + return _build_response(response=response) + + +async def asyncio( +*, client: Client) -> Optional[Union[Any, FileConversion]]: + """ Stop all async conversions that are currently running. This endpoint can only be used by specific KittyCAD employees. """ + + return (await asyncio_detailed( + client=client, + )).parsed diff --git a/kittycad/api/meta/__init__.py b/kittycad/api/meta/__init__.py index e69de29bb..8ab89b892 100644 --- a/kittycad/api/meta/__init__.py +++ b/kittycad/api/meta/__init__.py @@ -0,0 +1 @@ +""" Contains methods for accessing the meta API paths: Meta information about servers, instances, and sessions. """ diff --git a/kittycad/api/meta/auth_session.py b/kittycad/api/meta/auth_session.py new file mode 100644 index 000000000..48e3aa2b7 --- /dev/null +++ b/kittycad/api/meta/auth_session.py @@ -0,0 +1,88 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...models.auth_session import AuthSession +from ...types import Response + +def _get_kwargs( +*, client: Client) -> Dict[str, Any]: + url = "{}/_meta/debug/session".format(client.base_url, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, AuthSession]]: + if response.status_code == 200: + response_200 = AuthSession.from_dict(response.json()) + return response_200 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, AuthSession]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( +*, client: Client) -> Response[Union[Any, AuthSession]]: + kwargs = _get_kwargs( + client=client, + ) + + response = httpx.get( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( +*, client: Client) -> Optional[Union[Any, AuthSession]]: + """ Get information about your API request session. This is primarily used for debugging. """ + + return sync_detailed( + client=client, + ).parsed + + +async def asyncio_detailed( +*, client: Client) -> Response[Union[Any, AuthSession]]: + kwargs = _get_kwargs( + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.get(**kwargs) + + return _build_response(response=response) + + +async def asyncio( +*, client: Client) -> Optional[Union[Any, AuthSession]]: + """ Get information about your API request session. This is primarily used for debugging. """ + + return (await asyncio_detailed( + client=client, + )).parsed diff --git a/kittycad/api/meta/instance_metadata.py b/kittycad/api/meta/instance_metadata.py new file mode 100644 index 000000000..bba74ec86 --- /dev/null +++ b/kittycad/api/meta/instance_metadata.py @@ -0,0 +1,88 @@ +from typing import Any, Dict, Optional, Union + +import httpx + +from ...client import Client +from ...models.instance import Instance +from ...types import Response + +def _get_kwargs( +*, client: Client) -> Dict[str, Any]: + url = "{}/_meta/debug/instance".format(client.base_url, + ) + + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() + + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } + + +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, Instance]]: + if response.status_code == 200: + response_200 = Instance.from_dict(response.json()) + return response_200 + if response.status_code == 401: + response_401 = None + return response_401 + if response.status_code == 403: + response_403 = None + return response_403 + return None + + +def _build_response(*, response: httpx.Response) -> Response[Union[Any, Instance]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) + + +def sync_detailed( +*, client: Client) -> Response[Union[Any, Instance]]: + kwargs = _get_kwargs( + client=client, + ) + + response = httpx.get( + verify=client.verify_ssl, + **kwargs, + ) + + return _build_response(response=response) + + +def sync( +*, client: Client) -> Optional[Union[Any, Instance]]: + """ Get information about this specific API server instance. This is primarily used for debugging. """ + + return sync_detailed( + client=client, + ).parsed + + +async def asyncio_detailed( +*, client: Client) -> Response[Union[Any, Instance]]: + kwargs = _get_kwargs( + client=client, + ) + + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.get(**kwargs) + + return _build_response(response=response) + + +async def asyncio( +*, client: Client) -> Optional[Union[Any, Instance]]: + """ Get information about this specific API server instance. This is primarily used for debugging. """ + + return (await asyncio_detailed( + client=client, + )).parsed diff --git a/kittycad/api/meta/meta_debug_instance.py b/kittycad/api/meta/meta_debug_instance.py deleted file mode 100644 index 716d96181..000000000 --- a/kittycad/api/meta/meta_debug_instance.py +++ /dev/null @@ -1,107 +0,0 @@ -from typing import Any, Dict, Optional, Union - -import httpx - -from ...client import AuthenticatedClient -from ...models.instance_metadata import InstanceMetadata -from ...types import Response - - -def _get_kwargs( - *, - client: AuthenticatedClient, -) -> Dict[str, Any]: - url = "{}/_meta/debug/instance".format(client.base_url) - - headers: Dict[str, Any] = client.get_headers() - cookies: Dict[str, Any] = client.get_cookies() - - return { - "url": url, - "headers": headers, - "cookies": cookies, - "timeout": client.get_timeout(), - } - - -def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, InstanceMetadata]]: - if response.status_code == 200: - response_200 = InstanceMetadata.from_dict(response.json()) - - return response_200 - if response.status_code == 400: - response_400 = None - - return response_400 - if response.status_code == 401: - response_401 = None - - return response_401 - if response.status_code == 403: - response_403 = None - - return response_403 - return None - - -def _build_response(*, response: httpx.Response) -> Response[Union[Any, InstanceMetadata]]: - return Response( - status_code=response.status_code, - content=response.content, - headers=response.headers, - parsed=_parse_response(response=response), - ) - - -def sync_detailed( - *, - client: AuthenticatedClient, -) -> Response[Union[Any, InstanceMetadata]]: - kwargs = _get_kwargs( - client=client, - ) - - response = httpx.get( - verify=client.verify_ssl, - **kwargs, - ) - - return _build_response(response=response) - - -def sync( - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, InstanceMetadata]]: - """Get information about this specific API server instance. This is primarily used for debugging.""" - - return sync_detailed( - client=client, - ).parsed - - -async def asyncio_detailed( - *, - client: AuthenticatedClient, -) -> Response[Union[Any, InstanceMetadata]]: - kwargs = _get_kwargs( - client=client, - ) - - async with httpx.AsyncClient(verify=client.verify_ssl) as _client: - response = await _client.get(**kwargs) - - return _build_response(response=response) - - -async def asyncio( - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, InstanceMetadata]]: - """Get information about this specific API server instance. This is primarily used for debugging.""" - - return ( - await asyncio_detailed( - client=client, - ) - ).parsed diff --git a/kittycad/api/meta/meta_debug_session.py b/kittycad/api/meta/meta_debug_session.py deleted file mode 100644 index 24acd3a02..000000000 --- a/kittycad/api/meta/meta_debug_session.py +++ /dev/null @@ -1,107 +0,0 @@ -from typing import Any, Dict, Optional, Union - -import httpx - -from ...client import AuthenticatedClient -from ...models.auth_session import AuthSession -from ...types import Response - - -def _get_kwargs( - *, - client: AuthenticatedClient, -) -> Dict[str, Any]: - url = "{}/_meta/debug/session".format(client.base_url) - - headers: Dict[str, Any] = client.get_headers() - cookies: Dict[str, Any] = client.get_cookies() - - return { - "url": url, - "headers": headers, - "cookies": cookies, - "timeout": client.get_timeout(), - } - - -def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, AuthSession]]: - if response.status_code == 200: - response_200 = AuthSession.from_dict(response.json()) - - return response_200 - if response.status_code == 400: - response_400 = None - - return response_400 - if response.status_code == 401: - response_401 = None - - return response_401 - if response.status_code == 403: - response_403 = None - - return response_403 - return None - - -def _build_response(*, response: httpx.Response) -> Response[Union[Any, AuthSession]]: - return Response( - status_code=response.status_code, - content=response.content, - headers=response.headers, - parsed=_parse_response(response=response), - ) - - -def sync_detailed( - *, - client: AuthenticatedClient, -) -> Response[Union[Any, AuthSession]]: - kwargs = _get_kwargs( - client=client, - ) - - response = httpx.get( - verify=client.verify_ssl, - **kwargs, - ) - - return _build_response(response=response) - - -def sync( - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, AuthSession]]: - """Get information about your API request session. This is primarily used for debugging.""" - - return sync_detailed( - client=client, - ).parsed - - -async def asyncio_detailed( - *, - client: AuthenticatedClient, -) -> Response[Union[Any, AuthSession]]: - kwargs = _get_kwargs( - client=client, - ) - - async with httpx.AsyncClient(verify=client.verify_ssl) as _client: - response = await _client.get(**kwargs) - - return _build_response(response=response) - - -async def asyncio( - *, - client: AuthenticatedClient, -) -> Optional[Union[Any, AuthSession]]: - """Get information about your API request session. This is primarily used for debugging.""" - - return ( - await asyncio_detailed( - client=client, - ) - ).parsed diff --git a/kittycad/api/meta/ping.py b/kittycad/api/meta/ping.py index 5a8972476..e24df2d73 100644 --- a/kittycad/api/meta/ping.py +++ b/kittycad/api/meta/ping.py @@ -1,95 +1,82 @@ -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Union import httpx from ...client import Client -from ...models.message import Message +from ...models.pong_message import PongMessage from ...types import Response - def _get_kwargs( - *, - client: Client, -) -> Dict[str, Any]: - url = "{}/ping".format(client.base_url) +*, client: Client) -> Dict[str, Any]: + url = "{}/ping".format(client.base_url, + ) - headers: Dict[str, Any] = client.get_headers() - cookies: Dict[str, Any] = client.get_cookies() + headers: Dict[str, Any] = client.get_headers() + cookies: Dict[str, Any] = client.get_cookies() - return { - "url": url, - "headers": headers, - "cookies": cookies, - "timeout": client.get_timeout(), - } + return { + "url": url, + "headers": headers, + "cookies": cookies, + "timeout": client.get_timeout(), + } -def _parse_response(*, response: httpx.Response) -> Optional[Message]: - if response.status_code == 200: - response_200 = Message.from_dict(response.json()) - - return response_200 - return None +def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, PongMessage]]: + if response.status_code == 200: + response_200 = PongMessage.from_dict(response.json()) + return response_200 + return None -def _build_response(*, response: httpx.Response) -> Response[Message]: - return Response( - status_code=response.status_code, - content=response.content, - headers=response.headers, - parsed=_parse_response(response=response), - ) +def _build_response(*, response: httpx.Response) -> Response[Union[Any, PongMessage]]: + return Response( + status_code=response.status_code, + content=response.content, + headers=response.headers, + parsed=_parse_response(response=response), + ) def sync_detailed( - *, - client: Client, -) -> Response[Message]: - kwargs = _get_kwargs( - client=client, - ) +*, client: Client) -> Response[Union[Any, PongMessage]]: + kwargs = _get_kwargs( + client=client, + ) - response = httpx.get( - verify=client.verify_ssl, - **kwargs, - ) + response = httpx.get( + verify=client.verify_ssl, + **kwargs, + ) - return _build_response(response=response) + return _build_response(response=response) def sync( - *, - client: Client, -) -> Optional[Message]: - """Simple ping to the server.""" +*, client: Client) -> Optional[Union[Any, PongMessage]]: + """ Simple ping to the server. """ - return sync_detailed( - client=client, - ).parsed + return sync_detailed( + client=client, + ).parsed async def asyncio_detailed( - *, - client: Client, -) -> Response[Message]: - kwargs = _get_kwargs( - client=client, - ) +*, client: Client) -> Response[Union[Any, PongMessage]]: + kwargs = _get_kwargs( + client=client, + ) - async with httpx.AsyncClient(verify=client.verify_ssl) as _client: - response = await _client.get(**kwargs) + async with httpx.AsyncClient(verify=client.verify_ssl) as _client: + response = await _client.get(**kwargs) - return _build_response(response=response) + return _build_response(response=response) async def asyncio( - *, - client: Client, -) -> Optional[Message]: - """Simple ping to the server.""" +*, client: Client) -> Optional[Union[Any, PongMessage]]: + """ Simple ping to the server. """ - return ( - await asyncio_detailed( - client=client, - ) - ).parsed + return (await asyncio_detailed( + client=client, + )).parsed