openapi-python-client/openapi_python_client/templates/endpoint_module.py.jinja at main · e2b-dev/openapi-python-client · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
from http import HTTPStatus
from typing import Any, Optional, Union, cast

import httpx

from ...client import AuthenticatedClient, Client
from ...types import Response, UNSET
from ... import errors

{% for relative in endpoint.relative_imports | sort %}
{{ relative }}
{% endfor %}

{% from "endpoint_macros.py.jinja" import header_params, cookie_params, query_params,
    arguments, client, kwargs, parse_response, docstring, body_to_kwarg %}

{% set return_string = endpoint.response_type() %}
{% set parsed_responses = (endpoint.responses | length > 0) and return_string != "Any" %}

def _get_kwargs(
    {{ arguments(endpoint, include_client=False) | indent(4) }}
) -> dict[str, Any]:
    {{ header_params(endpoint) | indent(4) }}

    {{ cookie_params(endpoint) | indent(4) }}

    {{ query_params(endpoint) | indent(4) }}

    _kwargs: dict[str, Any] = {
        "method": "{{ endpoint.method }}",
        {% if endpoint.path_parameters %}
        "url": "{{ endpoint.path }}".format(
        {%- for parameter in endpoint.path_parameters -%}
        {{parameter.python_name}}={{parameter.python_name}},
        {%- endfor -%}
        ),
        {% else %}
        "url": "{{ endpoint.path }}",
        {% endif %}
        {% if endpoint.query_parameters %}
        "params": params,
        {% endif %}
        {% if endpoint.cookie_parameters %}
        "cookies": cookies,
        {% endif %}
    }

{% if endpoint.bodies | length > 1 %}
{% for body in endpoint.bodies %}
    if isinstance(body, {{body.prop.get_type_string() }}):
        {{ body_to_kwarg(body) | indent(8) }}
        headers["Content-Type"] = "{{ body.content_type }}"
{% endfor %}
{% elif endpoint.bodies | length == 1 %}
{% set body = endpoint.bodies[0] %}
    {{ body_to_kwarg(body) | indent(4) }}
    {% if body.content_type != "multipart/form-data" %}{# Need httpx to set the boundary automatically #}
    headers["Content-Type"] = "{{ body.content_type }}"
    {% endif %}
{% endif %}

{% if endpoint.header_parameters or endpoint.bodies | length > 0 %}
    _kwargs["headers"] = headers
{% endif %}
    return _kwargs


def _parse_response(*, client: Union[AuthenticatedClient, Client], response: httpx.Response) -> Optional[{{ return_string }}]:
    {% for response in endpoint.responses %}
    if response.status_code == {{ response.status_code.value }}:
        {% if parsed_responses %}{% import "property_templates/" + response.prop.template as prop_template %}
        {% if prop_template.construct %}
        {{ prop_template.construct(response.prop, response.source.attribute) | indent(8) }}
        {% elif response.source.return_type == response.prop.get_type_string()  %}
        {{ response.prop.python_name }} = {{ response.source.attribute }}
        {% else %}
        {{ response.prop.python_name }} = cast({{ response.prop.get_type_string() }}, {{ response.source.attribute }})
        {% endif %}
        return {{ response.prop.python_name }}
        {% else %}
        return None
        {% endif %}
    {% endfor %}
    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[{{ return_string }}]:
    return Response(
        status_code=HTTPStatus(response.status_code),
        content=response.content,
        headers=response.headers,
        parsed=_parse_response(client=client, response=response),
    )


def sync_detailed(
    {{ arguments(endpoint) | indent(4) }}
) -> Response[{{ return_string }}]:
    {{ docstring(endpoint, return_string, is_detailed=true) | indent(4) }}

    kwargs = _get_kwargs(
        {{ kwargs(endpoint, include_client=False) }}
    )

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

    return _build_response(client=client, response=response)

{% if parsed_responses %}
def sync(
    {{ arguments(endpoint) | indent(4) }}
) -> Optional[{{ return_string }}]:
    {{ docstring(endpoint, return_string, is_detailed=false) | indent(4) }}

    return sync_detailed(
        {{ kwargs(endpoint) }}
    ).parsed
{% endif %}

async def asyncio_detailed(
    {{ arguments(endpoint) | indent(4) }}
) -> Response[{{ return_string }}]:
    {{ docstring(endpoint, return_string, is_detailed=true) | indent(4) }}

    kwargs = _get_kwargs(
        {{ kwargs(endpoint, include_client=False) }}
    )

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

    return _build_response(client=client, response=response)

{% if parsed_responses %}
async def asyncio(
    {{ arguments(endpoint) | indent(4) }}
) -> Optional[{{ return_string }}]:
    {{ docstring(endpoint, return_string, is_detailed=false) | indent(4) }}

    return (await asyncio_detailed(
        {{ kwargs(endpoint) }}
    )).parsed
{% endif %}