openapi-python-client/openapi_python_client/templates/endpoint_module.py.jinja at main · MalteBecker/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
from typing import Any, Dict, List, Optional, Union, cast

import httpx

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

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

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

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

def _get_kwargs(
    {{ arguments(endpoint) | indent(4) }}
) -> Dict[str, Any]:
    url = "{}{{ endpoint.path }}".format(
        client.base_url
        {%- for parameter in endpoint.path_parameters.values() -%}
        ,{{parameter.name}}={{parameter.python_name}}
        {%- endfor -%}
    )

    headers: Dict[str, Any] = client.get_headers()
    cookies: Dict[str, Any] = client.get_cookies()

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

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

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

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

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

    return {
        "url": url,
        "headers": headers,
        "cookies": cookies,
        "timeout": client.get_timeout(),
        {% if endpoint.form_body_class %}
        "data": form_data.to_dict(),
        {% elif endpoint.multipart_body %}
        "files": {{ "multipart_" + endpoint.multipart_body.python_name }},
        {% elif endpoint.json_body %}
        "json": {{ "json_" + endpoint.json_body.python_name }},
        {% endif %}
        {% if endpoint.query_parameters %}
        "params": params,
        {% endif %}
    }


{% if parsed_responses %}
def _parse_response(*, response: httpx.Response) -> Optional[{{ return_string }}]:
    {% for response in endpoint.responses %}
    if response.status_code == {{ response.status_code }}:
        {% if response.prop.template %}
            {% from "property_templates/" + response.prop.template import construct %}
        {{ construct(response.prop, response.source) | indent(8) }}
        {% else %}
        {{ response.prop.python_name }} = {{ response.source }}
        {% endif %}
        return {{ response.prop.python_name }}
    {% endfor %}
    return None
{% endif %}


def _build_response(*, response: httpx.Response) -> Response[{{ return_string }}]:
    return Response(
        status_code=response.status_code,
        content=response.content,
        headers=response.headers,
        {% if parsed_responses %}
        parsed=_parse_response(response=response),
        {% else %}
        parsed=None,
        {% endif %}
    )


def sync_detailed(
    {{ arguments(endpoint) | indent(4) }}
) -> Response[{{ return_string }}]:
    kwargs = _get_kwargs(
        {{ kwargs(endpoint) }}
    )

    response = httpx.{{ endpoint.method }}(
        verify=client.verify_ssl,
        **kwargs,
    )

    return _build_response(response=response)

{% if parsed_responses %}
def sync(
    {{ arguments(endpoint) | indent(4) }}
) -> Optional[{{ return_string }}]:
    """ {{ endpoint.description }} """

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

async def asyncio_detailed(
    {{ arguments(endpoint) | indent(4) }}
) -> Response[{{ return_string }}]:
    kwargs = _get_kwargs(
        {{ kwargs(endpoint) }}
    )

    async with httpx.AsyncClient(verify=client.verify_ssl) as _client:
        response = await _client.{{ endpoint.method }}(
            **kwargs
        )

    return _build_response(response=response)

{% if parsed_responses %}
async def asyncio(
    {{ arguments(endpoint) | indent(4) }}
) -> Optional[{{ return_string }}]:
    """ {{ endpoint.description }} """

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