Skip to content

Commit

Permalink
feat: support prefixItems for arrays
Browse files Browse the repository at this point in the history
Generates a union of all types in `prefixItems` and `items` for the inner list item type
  • Loading branch information
estyrke committed Oct 15, 2024
1 parent b9694bd commit 23909aa
Show file tree
Hide file tree
Showing 8 changed files with 410 additions and 12 deletions.
31 changes: 31 additions & 0 deletions end_to_end_tests/3.1_specific.openapi.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -47,3 +47,34 @@ paths:
"application/json":
schema:
const: "Why have a fixed response? I dunno"
"/prefixItems":
post:
tags: [ "prefixItems" ]
requestBody:
required: true
content:
"application/json":
schema:
type: object
properties:
prefixItemsAndItems:
type: array
prefixItems:
- type: string
const: "prefix"
- type: string
items:
type: number
prefixItemsOnly:
type: array
prefixItems:
- type: string
- type: number
maxItems: 2
responses:
"200":
description: "Successful Response"
content:
"application/json":
schema:
type: string
Original file line number Diff line number Diff line change
@@ -0,0 +1,150 @@
from http import HTTPStatus
from typing import Any, Dict, Optional, Union, cast

import httpx

from ... import errors
from ...client import AuthenticatedClient, Client
from ...models.post_prefix_items_body import PostPrefixItemsBody
from ...types import Response


def _get_kwargs(
*,
body: PostPrefixItemsBody,
) -> Dict[str, Any]:
headers: Dict[str, Any] = {}

_kwargs: Dict[str, Any] = {
"method": "post",
"url": "/prefixItems",
}

_body = body.to_dict()

_kwargs["json"] = _body
headers["Content-Type"] = "application/json"

_kwargs["headers"] = headers
return _kwargs


def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[str]:
if response.status_code == HTTPStatus.OK:
response_200 = cast(str, response.json())
return response_200
if client.raise_on_unexpected_status:
raise errors.UnexpectedStatus(response.status_code, response.content)
else:
return None


def _build_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Response[str]:
return Response(
status_code=HTTPStatus(response.status_code),
content=response.content,
headers=response.headers,
parsed=_parse_response(client=client, response=response),
)


def sync_detailed(
*,
client: Union[AuthenticatedClient, Client],
body: PostPrefixItemsBody,
) -> Response[str]:
"""
Args:
body (PostPrefixItemsBody):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[str]
"""

kwargs = _get_kwargs(
body=body,
)

response = client.get_httpx_client().request(
**kwargs,
)

return _build_response(client=client, response=response)


def sync(
*,
client: Union[AuthenticatedClient, Client],
body: PostPrefixItemsBody,
) -> Optional[str]:
"""
Args:
body (PostPrefixItemsBody):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
str
"""

return sync_detailed(
client=client,
body=body,
).parsed


async def asyncio_detailed(
*,
client: Union[AuthenticatedClient, Client],
body: PostPrefixItemsBody,
) -> Response[str]:
"""
Args:
body (PostPrefixItemsBody):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
Response[str]
"""

kwargs = _get_kwargs(
body=body,
)

response = await client.get_async_httpx_client().request(**kwargs)

return _build_response(client=client, response=response)


async def asyncio(
*,
client: Union[AuthenticatedClient, Client],
body: PostPrefixItemsBody,
) -> Optional[str]:
"""
Args:
body (PostPrefixItemsBody):
Raises:
errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True.
httpx.TimeoutException: If the request takes longer than Client.timeout.
Returns:
str
"""

return (
await asyncio_detailed(
client=client,
body=body,
)
).parsed
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
"""Contains all the data models used in inputs/outputs"""

from .post_const_path_body import PostConstPathBody
from .post_prefix_items_body import PostPrefixItemsBody

__all__ = ("PostConstPathBody",)
__all__ = (
"PostConstPathBody",
"PostPrefixItemsBody",
)
Original file line number Diff line number Diff line change
@@ -0,0 +1,103 @@
from typing import Any, Dict, List, Literal, Type, TypeVar, Union, cast

from attrs import define as _attrs_define
from attrs import field as _attrs_field

from ..types import UNSET, Unset

T = TypeVar("T", bound="PostPrefixItemsBody")


@_attrs_define
class PostPrefixItemsBody:
"""
Attributes:
prefix_items_and_items (Union[Unset, List[Union[Literal['prefix'], float, str]]]):
prefix_items_only (Union[Unset, List[Union[float, str]]]):
"""

prefix_items_and_items: Union[Unset, List[Union[Literal["prefix"], float, str]]] = UNSET
prefix_items_only: Union[Unset, List[Union[float, str]]] = UNSET
additional_properties: Dict[str, Any] = _attrs_field(init=False, factory=dict)

def to_dict(self) -> Dict[str, Any]:
prefix_items_and_items: Union[Unset, List[Union[Literal["prefix"], float, str]]] = UNSET
if not isinstance(self.prefix_items_and_items, Unset):
prefix_items_and_items = []
for prefix_items_and_items_item_data in self.prefix_items_and_items:
prefix_items_and_items_item: Union[Literal["prefix"], float, str]
prefix_items_and_items_item = prefix_items_and_items_item_data
prefix_items_and_items.append(prefix_items_and_items_item)

prefix_items_only: Union[Unset, List[Union[float, str]]] = UNSET
if not isinstance(self.prefix_items_only, Unset):
prefix_items_only = []
for prefix_items_only_item_data in self.prefix_items_only:
prefix_items_only_item: Union[float, str]
prefix_items_only_item = prefix_items_only_item_data
prefix_items_only.append(prefix_items_only_item)

field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update({})
if prefix_items_and_items is not UNSET:
field_dict["prefixItemsAndItems"] = prefix_items_and_items
if prefix_items_only is not UNSET:
field_dict["prefixItemsOnly"] = prefix_items_only

return field_dict

@classmethod
def from_dict(cls: Type[T], src_dict: Dict[str, Any]) -> T:
d = src_dict.copy()
prefix_items_and_items = []
_prefix_items_and_items = d.pop("prefixItemsAndItems", UNSET)
for prefix_items_and_items_item_data in _prefix_items_and_items or []:

def _parse_prefix_items_and_items_item(data: object) -> Union[Literal["prefix"], float, str]:
prefix_items_and_items_item_type_0 = cast(Literal["prefix"], data)
if prefix_items_and_items_item_type_0 != "prefix":
raise ValueError(
f"prefixItemsAndItems_item_type_0 must match const 'prefix', got '{prefix_items_and_items_item_type_0}'"
)
return prefix_items_and_items_item_type_0
return cast(Union[Literal["prefix"], float, str], data)

prefix_items_and_items_item = _parse_prefix_items_and_items_item(prefix_items_and_items_item_data)

prefix_items_and_items.append(prefix_items_and_items_item)

prefix_items_only = []
_prefix_items_only = d.pop("prefixItemsOnly", UNSET)
for prefix_items_only_item_data in _prefix_items_only or []:

def _parse_prefix_items_only_item(data: object) -> Union[float, str]:
return cast(Union[float, str], data)

prefix_items_only_item = _parse_prefix_items_only_item(prefix_items_only_item_data)

prefix_items_only.append(prefix_items_only_item)

post_prefix_items_body = cls(
prefix_items_and_items=prefix_items_and_items,
prefix_items_only=prefix_items_only,
)

post_prefix_items_body.additional_properties = d
return post_prefix_items_body

@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())

def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]

def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value

def __delitem__(self, key: str) -> None:
del self.additional_properties[key]

def __contains__(self, key: str) -> bool:
return key in self.additional_properties
22 changes: 19 additions & 3 deletions openapi_python_client/parser/properties/list_property.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,12 +58,28 @@ def build(
"""
from . import property_from_data

if data.items is None:
return PropertyError(data=data, detail="type array must have items defined"), schemas
if data.items is None and not data.prefixItems:
return (
PropertyError(
data=data,
detail="type array must have items or prefixItems defined",
),
schemas,
)

items = data.prefixItems or []
if data.items:
items.append(data.items)

if len(items) == 1:
inner_schema = items[0]
else:
inner_schema = oai.Schema(anyOf=items)

inner_prop, schemas = property_from_data(
name=f"{name}_item",
required=True,
data=data.items,
data=inner_schema,
schemas=schemas,
parent_name=parent_name,
config=config,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ class Schema(BaseModel):
anyOf: List[Union[Reference, "Schema"]] = Field(default_factory=list)
schema_not: Optional[Union[Reference, "Schema"]] = Field(default=None, alias="not")
items: Optional[Union[Reference, "Schema"]] = None
prefixItems: Optional[List[Union[Reference, "Schema"]]] = Field(default_factory=list)
properties: Optional[Dict[str, Union[Reference, "Schema"]]] = None
additionalProperties: Optional[Union[bool, Reference, "Schema"]] = None
description: Optional[str] = None
Expand Down
Loading

0 comments on commit 23909aa

Please sign in to comment.