-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Compatible with openapi-python-client version 0.22.0. Adding only those which are already used in this repo for version 0.10.7 so far.
- Loading branch information
1 parent
9fce0ff
commit cd85079
Showing
12 changed files
with
1,054 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,170 @@ | ||
{# | ||
This is a custom template derived from: | ||
https://github.com/openapi-generators/openapi-python-client/tree/v0.22.0/openapi_python_client/templates/client.py.jinja | ||
#} | ||
import ssl | ||
from typing import Any, Union, Optional | ||
|
||
from attrs import define, field, evolve | ||
import httpx | ||
|
||
|
||
@define | ||
class Client: | ||
"""A class for keeping track of data related to the API | ||
|
||
{% macro httpx_args_docstring() %} | ||
The following are accepted as keyword arguments and will be used to construct httpx Clients internally: | ||
|
||
``base_url``: The base URL for the API, all requests are made to a relative path to this URL | ||
|
||
``cookies``: A dictionary of cookies to be sent with every request | ||
|
||
``headers``: A dictionary of headers to be sent with every request | ||
|
||
``timeout``: The maximum amount of a time a request can take. API functions will raise | ||
httpx.TimeoutException if this is exceeded. | ||
|
||
``verify_ssl``: Whether or not to verify the SSL certificate of the API server. This should be True in production, | ||
but can be set to False for testing purposes. | ||
|
||
``follow_redirects``: Whether or not to follow redirects. Default value is False. | ||
|
||
``httpx_args``: A dictionary of additional arguments to be passed to the ``httpx.Client`` and ``httpx.AsyncClient`` constructor. | ||
{% endmacro %} | ||
{{ httpx_args_docstring() }} | ||
|
||
Attributes: | ||
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a | ||
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword | ||
argument to the constructor. | ||
""" | ||
{% macro attributes() %} | ||
raise_on_unexpected_status: bool = field(default=False, kw_only=True) | ||
_base_url: str = field(alias="base_url") | ||
_cookies: dict[str, str] = field(factory=dict, kw_only=True, alias="cookies") | ||
_headers: dict[str, str] = field(factory=dict, kw_only=True, alias="headers") | ||
_timeout: Optional[httpx.Timeout] = field(default=None, kw_only=True, alias="timeout") | ||
_verify_ssl: Union[str, bool, ssl.SSLContext] = field(default=True, kw_only=True, alias="verify_ssl") | ||
_follow_redirects: bool = field(default=False, kw_only=True, alias="follow_redirects") | ||
_httpx_args: dict[str, Any] = field(factory=dict, kw_only=True, alias="httpx_args") | ||
_client: Optional[httpx.Client] = field(default=None, init=False) | ||
_async_client: Optional[httpx.AsyncClient] = field(default=None, init=False) | ||
{% endmacro %}{{ attributes() }} | ||
{% macro builders(self) %} | ||
def with_headers(self, headers: dict[str, str]) -> "{{ self }}": | ||
"""Get a new client matching this one with additional headers""" | ||
if self._client is not None: | ||
self._client.headers.update(headers) | ||
if self._async_client is not None: | ||
self._async_client.headers.update(headers) | ||
return evolve(self, headers={**self._headers, **headers}) | ||
|
||
def with_cookies(self, cookies: dict[str, str]) -> "{{ self }}": | ||
"""Get a new client matching this one with additional cookies""" | ||
if self._client is not None: | ||
self._client.cookies.update(cookies) | ||
if self._async_client is not None: | ||
self._async_client.cookies.update(cookies) | ||
return evolve(self, cookies={**self._cookies, **cookies}) | ||
|
||
def with_timeout(self, timeout: httpx.Timeout) -> "{{ self }}": | ||
"""Get a new client matching this one with a new timeout (in seconds)""" | ||
if self._client is not None: | ||
self._client.timeout = timeout | ||
if self._async_client is not None: | ||
self._async_client.timeout = timeout | ||
return evolve(self, timeout=timeout) | ||
{% endmacro %}{{ builders("Client") }} | ||
{% macro httpx_stuff(name, custom_constructor=None) %} | ||
def set_httpx_client(self, client: httpx.Client) -> "{{ name }}": | ||
"""Manually set the underlying httpx.Client | ||
|
||
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. | ||
""" | ||
self._client = client | ||
return self | ||
|
||
def get_httpx_client(self) -> httpx.Client: | ||
"""Get the underlying httpx.Client, constructing a new one if not previously set""" | ||
if self._client is None: | ||
{% if custom_constructor %} | ||
{{ custom_constructor | indent(12) }} | ||
{% endif %} | ||
self._client = httpx.Client( | ||
base_url=self._base_url, | ||
cookies=self._cookies, | ||
headers=self._headers, | ||
timeout=self._timeout, | ||
verify=self._verify_ssl, | ||
follow_redirects=self._follow_redirects, | ||
**self._httpx_args, | ||
) | ||
return self._client | ||
|
||
def __enter__(self) -> "{{ name }}": | ||
"""Enter a context manager for self.client—you cannot enter twice (see httpx docs)""" | ||
self.get_httpx_client().__enter__() | ||
return self | ||
|
||
def __exit__(self, *args: Any, **kwargs: Any) -> None: | ||
"""Exit a context manager for internal httpx.Client (see httpx docs)""" | ||
self.get_httpx_client().__exit__(*args, **kwargs) | ||
|
||
def set_async_httpx_client(self, async_client: httpx.AsyncClient) -> "{{ name }}": | ||
"""Manually the underlying httpx.AsyncClient | ||
|
||
**NOTE**: This will override any other settings on the client, including cookies, headers, and timeout. | ||
""" | ||
self._async_client = async_client | ||
return self | ||
|
||
def get_async_httpx_client(self) -> httpx.AsyncClient: | ||
"""Get the underlying httpx.AsyncClient, constructing a new one if not previously set""" | ||
if self._async_client is None: | ||
{% if custom_constructor %} | ||
{{ custom_constructor | indent(12) }} | ||
{% endif %} | ||
self._async_client = httpx.AsyncClient( | ||
base_url=self._base_url, | ||
cookies=self._cookies, | ||
headers=self._headers, | ||
timeout=self._timeout, | ||
verify=self._verify_ssl, | ||
follow_redirects=self._follow_redirects, | ||
**self._httpx_args, | ||
) | ||
return self._async_client | ||
|
||
async def __aenter__(self) -> "{{ name }}": | ||
"""Enter a context manager for underlying httpx.AsyncClient—you cannot enter twice (see httpx docs)""" | ||
await self.get_async_httpx_client().__aenter__() | ||
return self | ||
|
||
async def __aexit__(self, *args: Any, **kwargs: Any) -> None: | ||
"""Exit a context manager for underlying httpx.AsyncClient (see httpx docs)""" | ||
await self.get_async_httpx_client().__aexit__(*args, **kwargs) | ||
{% endmacro %}{{ httpx_stuff("Client") }} | ||
|
||
@define | ||
class AuthenticatedClient: | ||
"""A Client which has been authenticated for use on secured endpoints | ||
|
||
{{ httpx_args_docstring() }} | ||
|
||
Attributes: | ||
raise_on_unexpected_status: Whether or not to raise an errors.UnexpectedStatus if the API returns a | ||
status code that was not documented in the source OpenAPI document. Can also be provided as a keyword | ||
argument to the constructor. | ||
token: The token to use for authentication | ||
prefix: The prefix to use for the Authorization header | ||
auth_header_name: The name of the Authorization header | ||
""" | ||
|
||
{{ attributes() }} | ||
token: str | ||
prefix: str = "Bearer" | ||
auth_header_name: str = "Authorization" | ||
|
||
{{ builders("AuthenticatedClient") }} | ||
{{ httpx_stuff("AuthenticatedClient", "self._headers[self.auth_header_name] = f\"{self.prefix} {self.token}\" if self.prefix else self.token") }} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,4 @@ | ||
{# | ||
This is a custom template derived from: | ||
https://github.com/openapi-generators/openapi-python-client/tree/v0.22.0/openapi_python_client/templates/endpoint_init.py.jinja | ||
#} |
190 changes: 190 additions & 0 deletions
190
osidb_bindings/templates_0.22.0/endpoint_macros.py.jinja
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,190 @@ | ||
{# | ||
This is a custom template derived from: | ||
https://github.com/openapi-generators/openapi-python-client/blob/v0.22.0/openapi_python_client/templates/endpoint_macros.py.jinja | ||
#} | ||
{% from "property_templates/helpers.jinja" import guarded_statement %} | ||
{% from "helpers.jinja" import safe_docstring %} | ||
|
||
{% macro header_params(endpoint) %} | ||
{% if endpoint.header_parameters or endpoint.bodies | length > 0 %} | ||
headers: dict[str, Any] = {} | ||
{% if endpoint.header_parameters %} | ||
{% for parameter in endpoint.header_parameters %} | ||
{% import "property_templates/" + parameter.template as param_template %} | ||
{% if param_template.transform_header %} | ||
{% set expression = param_template.transform_header(parameter.python_name) %} | ||
{% else %} | ||
{% set expression = parameter.python_name %} | ||
{% endif %} | ||
{% set statement = 'headers["' + parameter.name + '"]' + " = " + expression %} | ||
{{ guarded_statement(parameter, parameter.python_name, statement) }} | ||
{% endfor %} | ||
{% endif %} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{% macro cookie_params(endpoint) %} | ||
{% if endpoint.cookie_parameters %} | ||
cookies = {} | ||
{% for parameter in endpoint.cookie_parameters %} | ||
{% if parameter.required %} | ||
cookies["{{ parameter.name}}"] = {{ parameter.python_name }} | ||
{% else %} | ||
if {{ parameter.python_name }} is not UNSET: | ||
cookies["{{ parameter.name}}"] = {{ parameter.python_name }} | ||
{% endif %} | ||
|
||
{% endfor %} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
|
||
{% macro query_params(endpoint) %} | ||
{% if endpoint.query_parameters %} | ||
params: dict[str, Any] = {} | ||
|
||
{% for property in endpoint.query_parameters %} | ||
{% set destination = property.python_name %} | ||
{% import "property_templates/" + property.template as prop_template %} | ||
{% if prop_template.transform %} | ||
{% set destination = "json_" + property.python_name %} | ||
{{ prop_template.transform(property, property.python_name, destination) }} | ||
{% endif %} | ||
{%- if not property.json_is_dict %} | ||
params["{{ property.name }}"] = {{ destination }} | ||
{% else %} | ||
{{ guarded_statement(property, destination, "params.update(" + destination + ")") }} | ||
{% endif %} | ||
|
||
{% endfor %} | ||
|
||
params = {k: v for k, v in params.items() if v is not UNSET and v is not None} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{% macro body_to_kwarg(body, destination) %} | ||
{% if body.body_type == "data" %} | ||
{{ destination }} = body.to_dict() | ||
{% elif body.body_type == "files"%} | ||
{{ multipart_body(body, destination) }} | ||
{% elif body.body_type == "json" %} | ||
{{ json_body(body, destination) }} | ||
{% elif body.body_type == "content" %} | ||
{{ destination }} = body.payload | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{% macro json_body(body, destination) %} | ||
{% set property = body.prop %} | ||
{% import "property_templates/" + property.template as prop_template %} | ||
{% if prop_template.transform %} | ||
{{ prop_template.transform(property, property.python_name, destination) }} | ||
{% else %} | ||
{{ destination }} = {{ property.python_name }} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{% macro multipart_body(body, destination) %} | ||
{% set property = body.prop %} | ||
{% import "property_templates/" + property.template as prop_template %} | ||
{% if prop_template.transform_multipart_body %} | ||
{{ prop_template.transform_multipart_body(property, property.python_name, destination) }} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{# The all the kwargs passed into an endpoint (and variants thereof)) #} | ||
{% macro arguments(endpoint, include_client=True) %} | ||
{# path parameters #} | ||
{% for parameter in endpoint.path_parameters %} | ||
{{ parameter.to_string() }}, | ||
{% endfor %} | ||
{% if include_client or ((endpoint.list_all_parameters() | length) > (endpoint.path_parameters | length)) %} | ||
*, | ||
{% endif %} | ||
{# Proper client based on whether or not the endpoint requires authentication #} | ||
{% if include_client %} | ||
{% if endpoint.requires_security %} | ||
client: AuthenticatedClient, | ||
{% else %} | ||
client: Union[AuthenticatedClient, Client], | ||
{% endif %} | ||
{% endif %} | ||
{# Any allowed bodies #} | ||
{% if endpoint.bodies | length == 1 %} | ||
body: {{ endpoint.bodies[0].prop.get_type_string() }}, | ||
{% elif endpoint.bodies | length > 1 %} | ||
body: Union[ | ||
{% for body in endpoint.bodies %} | ||
{{ body.prop.get_type_string() }}, | ||
{% endfor %} | ||
], | ||
{% endif %} | ||
{# query parameters #} | ||
{% for parameter in endpoint.query_parameters %} | ||
{{ parameter.to_string() }}, | ||
{% endfor %} | ||
{% for parameter in endpoint.header_parameters %} | ||
{{ parameter.to_string() }}, | ||
{% endfor %} | ||
{# cookie parameters #} | ||
{% for parameter in endpoint.cookie_parameters %} | ||
{{ parameter.to_string() }}, | ||
{% endfor %} | ||
{% endmacro %} | ||
|
||
{# Just lists all kwargs to endpoints as name=name for passing to other functions #} | ||
{% macro kwargs(endpoint, include_client=True) %} | ||
{% for parameter in endpoint.path_parameters %} | ||
{{ parameter.python_name }}={{ parameter.python_name }}, | ||
{% endfor %} | ||
{% if include_client %} | ||
client=client, | ||
{% endif %} | ||
{% if endpoint.bodies | length > 0 %} | ||
body=body, | ||
{% endif %} | ||
{% for parameter in endpoint.query_parameters %} | ||
{{ parameter.python_name }}={{ parameter.python_name }}, | ||
{% endfor %} | ||
{% for parameter in endpoint.header_parameters %} | ||
{{ parameter.python_name }}={{ parameter.python_name }}, | ||
{% endfor %} | ||
{% for parameter in endpoint.cookie_parameters %} | ||
{{ parameter.python_name }}={{ parameter.python_name }}, | ||
{% endfor %} | ||
{% endmacro %} | ||
|
||
{% macro docstring_content(endpoint, return_string, is_detailed) %} | ||
{% if endpoint.summary %}{{ endpoint.summary | wordwrap(100)}} | ||
|
||
{% endif -%} | ||
{%- if endpoint.description %} {{ endpoint.description | wordwrap(100) }} | ||
|
||
{% endif %} | ||
{% if not endpoint.summary and not endpoint.description %} | ||
{# Leave extra space so that Args or Returns isn't at the top #} | ||
|
||
{% endif %} | ||
{% set all_parameters = endpoint.list_all_parameters() %} | ||
{% if all_parameters %} | ||
Args: | ||
{% for parameter in all_parameters %} | ||
{{ parameter.to_docstring() | wordwrap(90) | indent(8) }} | ||
{% endfor %} | ||
|
||
{% endif %} | ||
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: | ||
{% if is_detailed %} | ||
Response[{{ return_string }}] | ||
{% else %} | ||
{{ return_string }} | ||
{% endif %} | ||
{% endmacro %} | ||
|
||
{% macro docstring(endpoint, return_string, is_detailed) %} | ||
{{ safe_docstring(docstring_content(endpoint, return_string, is_detailed)) }} | ||
{% endmacro %} |
Oops, something went wrong.