LukeParke/speakeasy-playground
1
0
Fork 0
mirror of https://github.com/LukeHagar/speakeasy-playground.git synced 2025-12-08 04:21:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

union array

Browse Source
This commit is contained in:
Mike Lueders
2024-03-21 16:38:48 -05:00
parent 90a646d891
commit ce42d22be9
20 changed files with 218 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
.speakeasy/gen.lock
View File

@@ -1,21 +1,22 @@
lockVersion: 2.0.0
id: f973dc21-d1bd-4e48-971a-a3918272099b
management:
docChecksum: 3851061474ad0b751f638e537d98fd25
docChecksum: 90fcc44f64e17c4c822732af1e41e532
docVersion: "1"
speakeasyVersion: internal
generationVersion: 2.275.2
releaseVersion: 0.0.1
configChecksum: 5be955b039369e57b218570e7f64d06f
speakeasyVersion: 1.219.1
generationVersion: 2.286.7
releaseVersion: 0.1.1
configChecksum: 8122f6908a2420616420e157d90ab645
features:
python:
constsAndDefaults: 0.1.2
core: 4.5.0
flattening: 2.81.1
globalServerURLs: 2.82.1
core: 4.5.1
getRequestBodies: 2.81.1
globalServerURLs: 2.82.2
responseFormat: 0.1.0
generatedFiles:
- src/shippo/sdkconfiguration.py
- src/shippo/sdk.py
- py.typed
- pylintrc
- setup.py
- src/shippo/__init__.py
@@ -25,11 +26,13 @@ generatedFiles:
- src/shippo/models/errors/sdkerror.py
- tests/helpers.py
- src/shippo/models/operations/example.py
- src/shippo/models/components/examplewithoneofarray.py
- src/shippo/models/__init__.py
- src/shippo/models/errors/__init__.py
- src/shippo/models/operations/__init__.py
- docs/models/operations/examplerequest.md
- src/shippo/models/components/__init__.py
- docs/models/operations/exampleresponse.md
- docs/models/components/examplewithoneofarray.md
- docs/sdks/shippo/README.md
- USAGE.md
- .gitattributes

3
.speakeasy/gen.yaml
View File

@@ -12,7 +12,7 @@ generation:
auth:
oAuth2ClientCredentialsEnabled: false
python:
version: 0.0.1
version: 0.1.1
additionalDependencies:
dependencies: {}
extraDependencies:
@@ -33,3 +33,4 @@ python:
maxMethodParams: 4
outputModelSuffix: output
packageName: shippo-api-client
responseFormat: envelope

45
README.md
View File

@@ -31,15 +31,22 @@ pip install git+<UNSET>.git
```python
import shippo
from shippo.models import components
s = shippo.Shippo()
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = s.example(results_per_page=904965)
res = s.example(req)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
<!-- End SDK Example Usage [usage] -->
@@ -64,21 +71,27 @@ Handling errors in this SDK should largely match your expectations. All operati
```python
import shippo
from shippo.models import errors
from shippo.models import components, errors
s = shippo.Shippo()
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = None
try:
res = s.example(results_per_page=904965)
res = s.example(req)
except errors.SDKError as e:
# handle exception
raise(e)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
<!-- End Error Handling [errors] -->
@@ -97,17 +110,24 @@ You can override the default server globally by passing a server index to the `s
```python
import shippo
from shippo.models import components
s = shippo.Shippo(
server_idx=0,
)
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = s.example(results_per_page=904965)
res = s.example(req)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
@@ -116,17 +136,24 @@ if res.status_code == 200:
The default server can also be overridden globally by passing a URL to the `server_url: str` optional parameter when initializing the SDK client instance. For example:
```python
import shippo
from shippo.models import components
s = shippo.Shippo(
server_url="https://example.com",
)
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = s.example(results_per_page=904965)
res = s.example(req)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
<!-- End Server Selection [server] -->

11
USAGE.md
View File

@@ -1,14 +1,21 @@
<!-- Start SDK Example Usage [usage] -->
```python
import shippo
from shippo.models import components
s = shippo.Shippo()
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = s.example(results_per_page=904965)
res = s.example(req)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
<!-- End SDK Example Usage [usage] -->

8
docs/models/components/examplewithoneofarray.md Normal file
View File

@@ -0,0 +1,8 @@
# ExampleWithOneOfArray
## Fields
| Field | Type | Required | Description |
| ------------------ | ------------------ | ------------------ | ------------------ |
| `one_of_array` | List[*int*] | :heavy_check_mark: | N/A |

8
docs/models/operations/examplerequest.md
View File

@@ -1,8 +0,0 @@
# ExampleRequest
## Fields
| Field | Type | Required | Description |
| -------------------------------------------------- | -------------------------------------------------- | -------------------------------------------------- | -------------------------------------------------- |
| `results_per_page` | *Optional[int]* | :heavy_minus_sign: | The number of results to return per page (max 100) |

17
docs/sdks/shippo/README.md
View File

@@ -13,22 +13,29 @@
```python
import shippo
from shippo.models import components
s = shippo.Shippo()
req = components.ExampleWithOneOfArray(
one_of_array=[
904965,
],
)
res = s.example(results_per_page=904965)
res = s.example(req)
if res.status_code == 200:
if res is not None:
# handle response
pass
```
### Parameters
| Parameter | Type | Required | Description |
| -------------------------------------------------- | -------------------------------------------------- | -------------------------------------------------- | -------------------------------------------------- |
| `results_per_page` | *Optional[int]* | :heavy_minus_sign: | The number of results to return per page (max 100) |
| Parameter | Type | Required | Description |
| ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ | ------------------------------------------------------------------------------------ |
| `request` | [components.ExampleWithOneOfArray](../../models/components/examplewithoneofarray.md) | :heavy_check_mark: | The request object to use for the request. |
### Response

31
openapi.yaml
View File

@@ -10,20 +10,27 @@ paths:
"/example":
get:
operationId: Example
parameters:
- $ref: "#/components/parameters/ResultsPerPageQueryParam"
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/ExampleWithOneOfArray'
responses:
'200':
description: example
components:
parameters:
ResultsPerPageQueryParam:
description: The number of results to return per page (max 100)
in: query
name: results_per_page
schema:
type: integer
format: int64
maximum: 100
default: 25
responses: {}
schemas:
ExampleWithOneOfArray:
type: object
properties:
one_of_array:
oneOf:
- type: array
items:
type: integer
- type: array
items:
type: string
required:
- one_of_array

1
py.typed Normal file
View File

@@ -0,0 +1 @@
# Marker file for PEP 561. The package enables type hints.

2
setup.py
View File

@@ -10,7 +10,7 @@ except FileNotFoundError:
setuptools.setup(
name="shippo-api-client",
version="0.0.1",
version="0.1.1",
author="Speakeasy",
description="Python Client SDK Generated by Speakeasy",
long_description=long_description,

1
src/shippo/_hooks/__init__.py
View File

@@ -2,3 +2,4 @@
from .sdkhooks import *
from .types import *
from .registration import *

13
src/shippo/_hooks/registration.py Normal file
View File

@@ -0,0 +1,13 @@
from .types import Hooks
# This file is only ever generated once on the first generation and then is free to be modified.
# Any hooks you wish to add should be registered in the init_hooks function. Feel free to define them
# in this file or in separate files in the hooks folder.
def init_hooks(hooks: Hooks):
# pylint: disable=unused-argument
"""Add hooks by calling hooks.register{sdk_init/before_request/after_success/after_error}Hook
with an instance of a hook that implements that specific Hook interface
Hooks are registered per SDK instance, and are valid for the lifetime of the SDK instance"""

12
src/shippo/_hooks/sdkhooks.py
View File

@@ -2,17 +2,17 @@
import requests
from .types import SDKInitHook, BeforeRequestContext, BeforeRequestHook, AfterSuccessContext, AfterSuccessHook, AfterErrorContext, AfterErrorHook, Hooks
from .registration import init_hooks
from typing import List, Optional, Tuple, Union
class SDKHooks(Hooks):
sdk_init_hooks: List[SDKInitHook] = []
before_request_hooks: List[BeforeRequestHook] = []
after_success_hooks: List[AfterSuccessHook] = []
after_error_hooks: List[AfterErrorHook] = []
def __init__(self):
pass
self.sdk_init_hooks: List[SDKInitHook] = []
self.before_request_hooks: List[BeforeRequestHook] = []
self.after_success_hooks: List[AfterSuccessHook] = []
self.after_error_hooks: List[AfterErrorHook] = []
init_hooks(self)
def register_sdk_init_hook(self, hook: SDKInitHook) -> None:
self.sdk_init_hooks.append(hook)

5
src/shippo/models/components/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from .examplewithoneofarray import *
__all__ = ["ExampleWithOneOfArray"]

15
src/shippo/models/components/examplewithoneofarray.py Normal file
View File

@@ -0,0 +1,15 @@
"""Code generated by Speakeasy (https://speakeasyapi.dev). DO NOT EDIT."""
from __future__ import annotations
import dataclasses
from dataclasses_json import Undefined, dataclass_json
from shippo import utils
from typing import List
@dataclass_json(undefined=Undefined.EXCLUDE)
@dataclasses.dataclass
class ExampleWithOneOfArray:
one_of_array: List[int] = dataclasses.field(metadata={'dataclasses_json': { 'letter_case': utils.get_field_name('one_of_array') }})

2
src/shippo/models/operations/__init__.py
View File

@@ -2,4 +2,4 @@
from .example import *
__all__ = ["ExampleRequest","ExampleResponse"]
__all__ = ["ExampleResponse"]

9
src/shippo/models/operations/example.py
View File

@@ -3,15 +3,6 @@
from __future__ import annotations
import dataclasses
import requests as requests_http
from typing import Optional
@dataclasses.dataclass
class ExampleRequest:
results_per_page: Optional[int] = dataclasses.field(default=25, metadata={'query_param': { 'field_name': 'results_per_page', 'style': 'form', 'explode': True }})
r"""The number of results to return per page (max 100)"""
@dataclasses.dataclass

55
src/shippo/sdk.py
View File

@@ -4,7 +4,7 @@ import requests as requests_http
from .sdkconfiguration import SDKConfiguration
from shippo import utils
from shippo._hooks import HookContext, SDKHooks
from shippo.models import errors, operations
from shippo.models import components, errors, operations
from typing import Dict, Optional
class Shippo:
@@ -12,14 +12,14 @@ class Shippo:
sdk_configuration: SDKConfiguration
def __init__(self,
server_idx: int = None,
server_url: str = None,
url_params: Dict[str, str] = None,
client: requests_http.Session = None,
retry_config: utils.RetryConfig = None
server_idx: Optional[int] = None,
server_url: Optional[str] = None,
url_params: Optional[Dict[str, str]] = None,
client: Optional[requests_http.Session] = None,
retry_config: Optional[utils.RetryConfig] = None
) -> None:
"""Instantiates the SDK configuring it with the provided parameters.
:param server_idx: The index of the server to use for all operations
:type server_idx: int
:param server_url: The server URL to use for all operations
@@ -33,12 +33,17 @@ class Shippo:
"""
if client is None:
client = requests_http.Session()
if server_url is not None:
if url_params is not None:
server_url = utils.template_url(server_url, url_params)
self.sdk_configuration = SDKConfiguration(client, None, server_url, server_idx, retry_config=retry_config)
self.sdk_configuration = SDKConfiguration(
client,
server_url,
server_idx,
retry_config=retry_config
)
hooks = SDKHooks()
@@ -48,34 +53,28 @@ class Shippo:
self.sdk_configuration.server_url = server_url
# pylint: disable=protected-access
self.sdk_configuration._hooks=hooks
def example(self, results_per_page: Optional[int] = None) -> operations.ExampleResponse:
self.sdk_configuration._hooks = hooks
def example(self, request: Optional[components.ExampleWithOneOfArray]) -> operations.ExampleResponse:
hook_ctx = HookContext(operation_id='Example', oauth2_scopes=[], security_source=None)
request = operations.ExampleRequest(
results_per_page=results_per_page,
)
base_url = utils.template_url(*self.sdk_configuration.get_server_details())
url = base_url + '/example'
headers = {}
query_params = utils.get_query_params(operations.ExampleRequest, request)
req_content_type, data, form = utils.serialize_request_body(request, Optional[components.ExampleWithOneOfArray], "request", False, True, 'json')
if req_content_type is not None and req_content_type not in ('multipart/form-data', 'multipart/mixed'):
headers['content-type'] = req_content_type
headers['Accept'] = '*/*'
headers['user-agent'] = self.sdk_configuration.user_agent
client = self.sdk_configuration.client
try:
req = self.sdk_configuration.get_hooks().before_request(
hook_ctx,
requests_http.Request('GET', url, params=query_params, headers=headers).prepare(),
requests_http.Request('GET', url, data=data, files=form, headers=headers).prepare(),
)
http_res = client.send(req)
except Exception as e:
@@ -92,15 +91,15 @@ class Shippo:
raise result
http_res = result
content_type = http_res.headers.get('Content-Type')
res = operations.ExampleResponse(status_code=http_res.status_code, content_type=content_type, raw_response=http_res)
res = operations.ExampleResponse(status_code=http_res.status_code, content_type=http_res.headers.get('Content-Type'), raw_response=http_res)
if http_res.status_code == 200:
pass
elif http_res.status_code >= 400 and http_res.status_code < 500 or http_res.status_code >= 500 and http_res.status_code < 600:
raise errors.SDKError('API error occurred', http_res.status_code, http_res.text, http_res)
else:
raise errors.SDKError('unknown status code received', http_res.status_code, http_res.text, http_res)
return res

6
src/shippo/sdkconfiguration.py
View File

@@ -21,9 +21,9 @@ class SDKConfiguration:
server_idx: int = 0
language: str = 'python'
openapi_doc_version: str = '1'
sdk_version: str = '0.0.1'
gen_version: str = '2.275.2'
user_agent: str = 'speakeasy-sdk/python 0.0.1 2.275.2 1 shippo-api-client'
sdk_version: str = '0.1.1'
gen_version: str = '2.286.7'
user_agent: str = 'speakeasy-sdk/python 0.1.1 2.286.7 1 shippo-api-client'
retry_config: RetryConfig = None
_hooks: SDKHooks = None

107
src/shippo/utils/utils.py
View File

@@ -4,7 +4,7 @@ import base64
import json
import re
import sys
from dataclasses import Field, dataclass, fields, is_dataclass, make_dataclass
from dataclasses import Field, fields, is_dataclass, make_dataclass
from datetime import date, datetime
from decimal import Decimal
from email.message import Message
@@ -14,30 +14,15 @@ from typing import (Any, Callable, Dict, List, Optional, Tuple, Union,
from xmlrpc.client import boolean
from typing_inspect import is_optional_type
import dateutil.parser
import requests
from dataclasses_json import DataClassJsonMixin
class SecurityClient:
client: requests.Session
query_params: Dict[str, str] = {}
def get_security(security: Any) -> Tuple[Dict[str, str], Dict[str, str]]:
headers: Dict[str, str] = {}
def __init__(self, client: requests.Session):
self.client = client
def send(self, request: requests.PreparedRequest, **kwargs):
request.prepare_url(url=request.url, params=self.query_params)
request.headers.update(self.headers)
return self.client.send(request, **kwargs)
def configure_security_client(client: requests.Session, security: dataclass):
client = SecurityClient(client)
query_params: Dict[str, str] = {}
if security is None:
return client
return headers, query_params
sec_fields: Tuple[Field, ...] = fields(security)
for sec_field in sec_fields:
@@ -49,35 +34,35 @@ def configure_security_client(client: requests.Session, security: dataclass):
if metadata is None:
continue
if metadata.get('option'):
_parse_security_option(client, value)
return client
_parse_security_option(headers, query_params, value)
return headers, query_params
if metadata.get('scheme'):
# Special case for basic auth which could be a flattened struct
if metadata.get("sub_type") == "basic" and not is_dataclass(value):
_parse_security_scheme(client, metadata, security)
_parse_security_scheme(headers, query_params, metadata, security)
else:
_parse_security_scheme(client, metadata, value)
_parse_security_scheme(headers, query_params, metadata, value)
return client
return headers, query_params
def _parse_security_option(client: SecurityClient, option: dataclass):
def _parse_security_option(headers: Dict[str, str], query_params: Dict[str, str], option: Any):
opt_fields: Tuple[Field, ...] = fields(option)
for opt_field in opt_fields:
metadata = opt_field.metadata.get('security')
if metadata is None or metadata.get('scheme') is None:
continue
_parse_security_scheme(
client, metadata, getattr(option, opt_field.name))
headers, query_params, metadata, getattr(option, opt_field.name))
def _parse_security_scheme(client: SecurityClient, scheme_metadata: Dict, scheme: any):
def _parse_security_scheme(headers: Dict[str, str], query_params: Dict[str, str], scheme_metadata: Dict, scheme: Any):
scheme_type = scheme_metadata.get('type')
sub_type = scheme_metadata.get('sub_type')
if is_dataclass(scheme):
if scheme_type == 'http' and sub_type == 'basic':
_parse_basic_auth_scheme(client, scheme)
_parse_basic_auth_scheme(headers, scheme)
return
scheme_fields: Tuple[Field, ...] = fields(scheme)
@@ -89,33 +74,33 @@ def _parse_security_scheme(client: SecurityClient, scheme_metadata: Dict, scheme
value = getattr(scheme, scheme_field.name)
_parse_security_scheme_value(
client, scheme_metadata, metadata, value)
headers, query_params, scheme_metadata, metadata, value)
else:
_parse_security_scheme_value(
client, scheme_metadata, scheme_metadata, scheme)
headers, query_params, scheme_metadata, scheme_metadata, scheme)
def _parse_security_scheme_value(client: SecurityClient, scheme_metadata: Dict, security_metadata: Dict, value: any):
def _parse_security_scheme_value(headers: Dict[str, str], query_params: Dict[str, str], scheme_metadata: Dict, security_metadata: Dict, value: Any):
scheme_type = scheme_metadata.get('type')
sub_type = scheme_metadata.get('sub_type')
header_name = security_metadata.get('field_name')
header_name = str(security_metadata.get('field_name'))
if scheme_type == "apiKey":
if sub_type == 'header':
client.headers[header_name] = value
headers[header_name] = value
elif sub_type == 'query':
client.query_params[header_name] = value
query_params[header_name] = value
else:
raise Exception('not supported')
elif scheme_type == "openIdConnect":
client.headers[header_name] = _apply_bearer(value)
headers[header_name] = _apply_bearer(value)
elif scheme_type == 'oauth2':
if sub_type != 'client_credentials':
client.headers[header_name] = _apply_bearer(value)
headers[header_name] = _apply_bearer(value)
elif scheme_type == 'http':
if sub_type == 'bearer':
client.headers[header_name] = _apply_bearer(value)
headers[header_name] = _apply_bearer(value)
else:
raise Exception('not supported')
else:
@@ -126,7 +111,7 @@ def _apply_bearer(token: str) -> str:
return token.lower().startswith('bearer ') and token or f'Bearer {token}'
def _parse_basic_auth_scheme(client: SecurityClient, scheme: dataclass):
def _parse_basic_auth_scheme(headers: Dict[str, str], scheme: Any):
username = ""
password = ""
@@ -145,11 +130,11 @@ def _parse_basic_auth_scheme(client: SecurityClient, scheme: dataclass):
password = value
data = f'{username}:{password}'.encode()
client.headers['Authorization'] = f'Basic {base64.b64encode(data).decode()}'
headers['Authorization'] = f'Basic {base64.b64encode(data).decode()}'
def generate_url(clazz: type, server_url: str, path: str, path_params: dataclass,
gbls: Dict[str, Dict[str, Dict[str, Any]]] = None) -> str:
def generate_url(clazz: type, server_url: str, path: str, path_params: Any,
gbls: Optional[Dict[str, Dict[str, Dict[str, Any]]]] = None) -> str:
path_param_fields: Tuple[Field, ...] = fields(clazz)
for field in path_param_fields:
request_metadata = field.metadata.get('request')
@@ -241,7 +226,7 @@ def template_url(url_with_params: str, params: Dict[str, str]) -> str:
return url_with_params
def get_query_params(clazz: type, query_params: dataclass, gbls: Dict[str, Dict[str, Dict[str, Any]]] = None) -> Dict[
def get_query_params(clazz: type, query_params: Any, gbls: Optional[Dict[str, Dict[str, Dict[str, Any]]]] = None) -> Dict[
str, List[str]]:
params: Dict[str, List[str]] = {}
@@ -287,7 +272,7 @@ def get_query_params(clazz: type, query_params: dataclass, gbls: Dict[str, Dict[
return params
def get_headers(headers_params: dataclass) -> Dict[str, str]:
def get_headers(headers_params: Any) -> Dict[str, str]:
if headers_params is None:
return {}
@@ -308,7 +293,7 @@ def get_headers(headers_params: dataclass) -> Dict[str, str]:
return headers
def _get_serialized_params(metadata: Dict, field_type: type, field_name: str, obj: any) -> Dict[str, str]:
def _get_serialized_params(metadata: Dict, field_type: type, field_name: str, obj: Any) -> Dict[str, str]:
params: Dict[str, str] = {}
serialization = metadata.get('serialization', '')
@@ -319,7 +304,7 @@ def _get_serialized_params(metadata: Dict, field_type: type, field_name: str, ob
return params
def _get_deep_object_query_params(metadata: Dict, field_name: str, obj: any) -> Dict[str, List[str]]:
def _get_deep_object_query_params(metadata: Dict, field_name: str, obj: Any) -> Dict[str, List[str]]:
params: Dict[str, List[str]] = {}
if obj is None:
@@ -385,7 +370,7 @@ def _get_query_param_field_name(obj_field: Field) -> str:
return obj_param_metadata.get("field_name", obj_field.name)
def _get_delimited_query_params(metadata: Dict, field_name: str, obj: any, delimiter: str) -> Dict[
def _get_delimited_query_params(metadata: Dict, field_name: str, obj: Any, delimiter: str) -> Dict[
str, List[str]]:
return _populate_form(field_name, metadata.get("explode", True), obj, _get_query_param_field_name, delimiter)
@@ -399,8 +384,8 @@ SERIALIZATION_METHOD_TO_CONTENT_TYPE = {
}
def serialize_request_body(request: dataclass, request_type: type, request_field_name: str, nullable: bool, optional: bool, serialization_method: str, encoder=None) -> Tuple[
str, any, any]:
def serialize_request_body(request: Any, request_type: type, request_field_name: str, nullable: bool, optional: bool, serialization_method: str, encoder=None) -> Tuple[
Optional[str], Optional[Any], Optional[Any]]:
if request is None:
if not nullable and optional:
return None, None, None
@@ -430,7 +415,7 @@ def serialize_request_body(request: dataclass, request_type: type, request_field
request_val)
def serialize_content_type(field_name: str, request_type: any, media_type: str, request: dataclass, encoder=None) -> Tuple[str, any, List[List[any]]]:
def serialize_content_type(field_name: str, request_type: Any, media_type: str, request: Any, encoder=None) -> Tuple[Optional[str], Optional[Any], Optional[List[List[Any]]]]:
if re.match(r'(application|text)\/.*?\+*json.*', media_type) is not None:
return media_type, marshal_json(request, request_type, encoder), None
if re.match(r'multipart\/.*', media_type) is not None:
@@ -446,8 +431,8 @@ def serialize_content_type(field_name: str, request_type: any, media_type: str,
f"invalid request body type {type(request)} for mediaType {media_type}")
def serialize_multipart_form(media_type: str, request: dataclass) -> Tuple[str, any, List[List[any]]]:
form: List[List[any]] = []
def serialize_multipart_form(media_type: str, request: Any) -> Tuple[str, Any, List[List[Any]]]:
form: List[List[Any]] = []
request_fields = fields(request)
for field in request_fields:
@@ -502,7 +487,7 @@ def serialize_multipart_form(media_type: str, request: dataclass) -> Tuple[str,
def serialize_dict(original: Dict, explode: bool, field_name, existing: Optional[Dict[str, List[str]]]) -> Dict[
str, List[str]]:
if existing is None:
existing = []
existing = {}
if explode is True:
for key, val in original.items():
@@ -520,7 +505,7 @@ def serialize_dict(original: Dict, explode: bool, field_name, existing: Optional
return existing
def serialize_form_data(field_name: str, data: dataclass) -> Dict[str, any]:
def serialize_form_data(field_name: str, data: Any) -> Dict[str, Any]:
form: Dict[str, List[str]] = {}
if is_dataclass(data):
@@ -562,7 +547,7 @@ def _get_form_field_name(obj_field: Field) -> str:
return obj_param_metadata.get("field_name", obj_field.name)
def _populate_form(field_name: str, explode: boolean, obj: any, get_field_name_func: Callable, delimiter: str) -> \
def _populate_form(field_name: str, explode: boolean, obj: Any, get_field_name_func: Callable, delimiter: str) -> \
Dict[str, List[str]]:
params: Dict[str, List[str]] = {}
@@ -597,7 +582,7 @@ def _populate_form(field_name: str, explode: boolean, obj: any, get_field_name_f
continue
if explode:
params[key] = _val_to_string(value)
params[key] = [_val_to_string(value)]
else:
items.append(f'{key}{delimiter}{_val_to_string(value)}')
@@ -626,7 +611,7 @@ def _populate_form(field_name: str, explode: boolean, obj: any, get_field_name_f
return params
def _serialize_header(explode: bool, obj: any) -> str:
def _serialize_header(explode: bool, obj: Any) -> str:
if obj is None:
return ''
@@ -850,7 +835,7 @@ def list_decoder(value_decoder: Callable):
def union_encoder(all_encoders: Dict[str, Callable]):
def selective_encoder(val: any):
def selective_encoder(val: Any):
if type(val) in all_encoders:
return all_encoders[type(val)](val)
return val
@@ -858,7 +843,7 @@ def union_encoder(all_encoders: Dict[str, Callable]):
def union_decoder(all_decoders: List[Callable]):
def selective_decoder(val: any):
def selective_decoder(val: Any):
decoded = val
for decoder in all_decoders:
try:
@@ -877,18 +862,18 @@ def get_field_name(name):
return override
def _val_to_string(val):
def _val_to_string(val) -> str:
if isinstance(val, bool):
return str(val).lower()
if isinstance(val, datetime):
return val.isoformat().replace('+00:00', 'Z')
return str(val.isoformat().replace('+00:00', 'Z'))
if isinstance(val, Enum):
return str(val.value)
return str(val)
def _populate_from_globals(param_name: str, value: any, param_type: str, gbls: Dict[str, Dict[str, Dict[str, Any]]]):
def _populate_from_globals(param_name: str, value: Any, param_type: str, gbls: Optional[Dict[str, Dict[str, Dict[str, Any]]]]):
if value is None and gbls is not None:
if 'parameters' in gbls:
if param_type in gbls['parameters']: