mirror of
https://github.com/ansible-collections/community.general.git
synced 2026-03-22 05:09:12 +00:00
25c475a7ef
replace literal HTTP codes with `http.HTTPStatus` (#11561)
* replace literal HTTP codes with http.HTTPStatus
* add changelog frag
(cherry picked from commit 7436c0c9ba)
Co-authored-by: Alexei Znamensky <103110+russoz@users.noreply.github.com>
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
# Copyright (c) 2021, Phillipe Smith <phsmithcc@gmail.com>
|
|
# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
|
|
# SPDX-License-Identifier: GPL-3.0-or-later
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import traceback
|
|
import typing as t
|
|
from http import HTTPStatus
|
|
|
|
from ansible.module_utils.urls import fetch_url, url_argument_spec
|
|
|
|
if t.TYPE_CHECKING:
|
|
from ansible.module_utils.basic import AnsibleModule
|
|
|
|
|
|
def api_argument_spec() -> dict[str, t.Any]:
|
|
"""
|
|
Creates an argument spec that can be used with any module
|
|
that will be requesting content via Rundeck API
|
|
"""
|
|
api_argument_spec = url_argument_spec()
|
|
api_argument_spec.update(
|
|
dict(
|
|
url=dict(required=True, type="str"),
|
|
api_version=dict(type="int", default=39),
|
|
api_token=dict(required=True, type="str", no_log=True),
|
|
)
|
|
)
|
|
|
|
return api_argument_spec
|
|
|
|
|
|
def api_request(
|
|
module: AnsibleModule,
|
|
endpoint: str,
|
|
data: t.Any | None = None,
|
|
method: str = "GET",
|
|
content_type: str = "application/json",
|
|
) -> tuple[t.Any, dict[str, t.Any]]:
|
|
"""Manages Rundeck API requests via HTTP(S)
|
|
|
|
:arg module: The AnsibleModule (used to get url, api_version, api_token, etc).
|
|
:arg endpoint: The API endpoint to be used.
|
|
:kwarg data: The data to be sent (in case of POST/PUT).
|
|
:kwarg method: "POST", "PUT", etc.
|
|
|
|
:returns: A tuple of (**response**, **info**). Use ``response.read()`` to read the data.
|
|
The **info** contains the 'status' and other meta data. When a HttpError (status >= 400)
|
|
occurred then ``info['body']`` contains the error response data::
|
|
|
|
Example::
|
|
|
|
data={...}
|
|
resp, info = fetch_url(module,
|
|
"http://rundeck.example.org",
|
|
data=module.jsonify(data),
|
|
method="POST")
|
|
status_code = info["status"]
|
|
body = resp.read()
|
|
if status_code >= 400 :
|
|
body = info['body']
|
|
"""
|
|
|
|
response, info = fetch_url(
|
|
module=module,
|
|
url=f"{module.params['url']}/api/{module.params['api_version']}/{endpoint}",
|
|
data=json.dumps(data),
|
|
method=method,
|
|
headers={
|
|
"Content-Type": content_type,
|
|
"Accept": "application/json",
|
|
"X-Rundeck-Auth-Token": module.params["api_token"],
|
|
},
|
|
)
|
|
|
|
_status = HTTPStatus(info["status"])
|
|
|
|
if _status == HTTPStatus.FORBIDDEN:
|
|
module.fail_json(msg="Token authorization failed", execution_info=json.loads(info["body"]))
|
|
elif _status == HTTPStatus.NOT_FOUND:
|
|
return None, info
|
|
elif _status == HTTPStatus.CONFLICT:
|
|
module.fail_json(msg="Job executions limit reached", execution_info=json.loads(info["body"]))
|
|
elif _status.is_server_error:
|
|
module.fail_json(msg="Rundeck API error", execution_info=json.loads(info["body"]))
|
|
|
|
try:
|
|
content = response.read()
|
|
|
|
if not content:
|
|
return None, info
|
|
else:
|
|
json_response = json.loads(content)
|
|
return json_response, info
|
|
except AttributeError as error:
|
|
module.fail_json(
|
|
msg=f"Rundeck API request error: {error}", exception=traceback.format_exc(), execution_info=info
|
|
)
|
|
except ValueError as error:
|
|
module.fail_json(
|
|
msg=f"No valid JSON response: {error}", exception=traceback.format_exc(), execution_info=content
|
|
)
|