Signed-off-by: Jess Frazelle <github@jessfraz.com>
This commit is contained in:
Jess Frazelle
2022-06-11 17:59:55 -07:00
parent 49b8d2747a
commit 7c4321306a
9 changed files with 202 additions and 16 deletions

View File

@ -289,12 +289,33 @@ response: Response[""" + success_type + """] = await """ + fn_name + """.asyncio
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")
schema = data['components']['schemas'][ref]
# Let's check if it is a oneOf.
if 'oneOf' in schema:
# We want to parse each of the possible types.
for index, one_of in enumerate(schema['oneOf']):
ref = one_of['$ref'].replace(
'#/components/schemas/', '')
f.write("\t\ttry:\n")
f.write(
"\t\t\tif not isinstance(data, dict):\n")
f.write("\t\t\t\traise TypeError()\n")
f.write(
"\t\t\toption = " + ref + ".from_dict(data)\n")
f.write("\t\t\treturn option\n")
f.write("\t\texcept:\n")
if index == len(schema['oneOf']) - 1:
# On the last one raise the error.
f.write("\t\t\traise\n")
else:
f.write("\t\t\tpass\n")
else:
f.write(
"\t\tresponse_" +
response_code +
" = " +
ref +
".from_dict(response.json())\n")
elif 'type' in json:
if json['type'] == 'array':
items = json['items']

View File

@ -5,6 +5,8 @@ set -o pipefail
# Cleanup old stuff.
rm -rf kittycad/models
rm -rf kittycad/api
git checkout kittycad/api/file/create_file_conversion_with_base64_helper.py
git checkout kittycad/api/file/get_file_conversion_with_base64_helper.py
# Generate new.
poetry run python generate/generate.py

View File

@ -29,7 +29,27 @@ def _get_kwargs(
def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion, FileMass, FileVolume, Error]]:
if response.status_code == 200:
response_200 = AsyncApiCallOutput.from_dict(response.json())
try:
if not isinstance(data, dict):
raise TypeError()
option = FileConversion.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileMass.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileVolume.from_dict(data)
return option
except:
raise
return response_200
if response.status_code == 400:
response_4XX = Error.from_dict(response.json())

View File

@ -0,0 +1,59 @@
from typing import Any, Dict, Optional, Union
import base64
import httpx
from ...client import Client
from ...models import Error
from ...models import FileConversion
from ...models import FileSourceFormat
from ...models import FileOutputFormat
from ...types import Response
from ...api.file.create_file_conversion import sync as fc_sync, asyncio as fc_asyncio
def sync(
src_format: FileSourceFormat,
output_format: FileOutputFormat,
body: bytes,
*,
client: Client,
) -> Optional[Union[Any, FileConversion, Error]]:
"""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. This function automatically base64 encodes the request body and base64 decodes the request output."""
encoded = base64.b64encode(body)
fc = fc_sync(
src_format=src_format,
output_format=output_format,
body=encoded,
client=client,
)
if isinstance(fc, FileConversion) and fc.output != "":
fc.output = base64.b64decode(fc.output)
return fc
async def asyncio(
src_format: FileSourceFormat,
output_format: FileOutputFormat,
body: bytes,
*,
client: Client,
) -> Optional[Union[Any, FileConversion, Error]]:
"""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. This function automatically base64 encodes the request body and base64 decodes the request output."""
encoded = base64.b64encode(body)
fc = await fc_asyncio(
src_format=src_format,
output_format=output_format,
body=encoded,
client=client,
)
if isinstance(fc, FileConversion) and fc.output != "":
fc.output = base64.b64decode(fc.output)
return fc

View File

@ -29,7 +29,27 @@ def _get_kwargs(
def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion, FileMass, FileVolume, Error]]:
if response.status_code == 200:
response_200 = AsyncApiCallOutput.from_dict(response.json())
try:
if not isinstance(data, dict):
raise TypeError()
option = FileConversion.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileMass.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileVolume.from_dict(data)
return option
except:
raise
return response_200
if response.status_code == 400:
response_4XX = Error.from_dict(response.json())

View File

@ -29,7 +29,27 @@ def _get_kwargs(
def _parse_response(*, response: httpx.Response) -> Optional[Union[Any, FileConversion, FileMass, FileVolume, Error]]:
if response.status_code == 200:
response_200 = AsyncApiCallOutput.from_dict(response.json())
try:
if not isinstance(data, dict):
raise TypeError()
option = FileConversion.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileMass.from_dict(data)
return option
except:
pass
try:
if not isinstance(data, dict):
raise TypeError()
option = FileVolume.from_dict(data)
return option
except:
raise
return response_200
if response.status_code == 400:
response_4XX = Error.from_dict(response.json())

View File

@ -0,0 +1,46 @@
from typing import Any, Dict, Optional, Union
import base64
import httpx
from ...client import Client
from ...models import Error
from ...models.file_conversion import FileConversion
from ...types import Response
from ...api.file.get_file_conversion import sync as fc_sync, asyncio as fc_asyncio
def sync(
id: str,
*,
client: Client,
) -> Optional[Union[Any, FileConversion, Error]]:
"""Get the status of a file conversion. This function automatically base64 decodes the output response if there is one."""
fc = fc_sync(
id=id,
client=client,
)
if isinstance(fc, FileConversion) and fc.output != "":
fc.output = base64.b64decode(fc.output)
return fc
async def asyncio(
id: str,
*,
client: Client,
) -> Optional[Union[Any, FileConversion, Error]]:
"""Get the status of a file conversion. This function automatically base64 decodes the output response if there is one."""
fc = await fc_asyncio(
id=id,
client=client,
)
if isinstance(fc, FileConversion) and fc.output != "":
fc.output = base64.b64decode(fc.output)
return fc

View File

@ -3,7 +3,7 @@ import pytest
import asyncio
from .client import ClientFromEnv
from .models import FileConversion, FileConversionOutputFormat, FileConversionSourceFormat, User, Pong, FileConversionStatus
from .models import FileConversion, FileOutputFormat, FileSourceFormat, User, Pong, APICallStatus
from .api.file import create_file_conversion_with_base64_helper
from .api.meta import ping
from .api.users import get_user_self
@ -72,8 +72,8 @@ def test_file_convert_stl():
fc: FileConversion = create_file_conversion_with_base64_helper.sync(
client=client,
body=content,
src_format=FileConversionSourceFormat.STL,
output_format=FileConversionOutputFormat.OBJ)
src_format=FileSourceFormat.STL,
output_format=FileOutputFormat.OBJ)
assert fc is not None
@ -81,7 +81,7 @@ def test_file_convert_stl():
assert fc.id is not None
assert fc.status == FileConversionStatus.COMPLETED
assert fc.status == APICallStatus.COMPLETED
print(f"FileConversion: {fc}")
@ -97,7 +97,7 @@ async def test_file_convert_stl_async():
file.close()
# Get the fc.
fc: FileConversion = await create_file_conversion_with_base64_helper.asyncio(client=client, body=content, src_format=FileConversionSourceFormat.STL, output_format=FileConversionOutputFormat.OBJ)
fc: FileConversion = await create_file_conversion_with_base64_helper.asyncio(client=client, body=content, src_format=FileSourceFormat.STL, output_format=FileOutputFormat.OBJ)
assert fc is not None

View File

@ -25,5 +25,3 @@ class Invoice:
id: Union[Unset, str] = UNSET
invoice_pdf: Union[Unset, str] = UNSET
invoice_url: Union[Unset, str] = UNSET
from ..models import InvoiceLineItem
lines: Union[Unset, List[InvoiceLineItem]] = UNSET