ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2026-03-22 05:09:12 +00:00
Code Activity

[PR #11561/7436c0c9 backport][stable-12] replace literal HTTP codes with http.HTTPStatus (#11568)

Browse source 
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>
This commit is contained in:
patchback[bot] 2026-03-10 22:14:27 +01:00 • committed by GitHub
parent b3782a76e0
commit 25c475a7ef
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
18 changed files with 135 additions and 78 deletions
Download patch file Download diff file Expand all files Collapse all files

11
plugins/inventory/icinga2.py
View file

@ -93,6 +93,7 @@ compose:
"""
import json
from http import HTTPStatus
from urllib.error import HTTPError
from ansible.errors import AnsibleParserError
@ -164,7 +165,7 @@ class InventoryModule(BaseInventoryPlugin, Constructable):
self.display.vvv(f"Error returned: {error_body}")
except Exception:
error_body = {"status": None}
if e.code == 404 and error_body.get("status") == "No objects found.":
if e.code == HTTPStatus.NOT_FOUND and error_body.get("status") == "No objects found.":
raise AnsibleParserError(
"Host filter returned no data. Please confirm your host_filter value is valid"
) from e
@ -173,15 +174,15 @@ class InventoryModule(BaseInventoryPlugin, Constructable):
response_body = response.read()
json_data = json.loads(response_body.decode("utf-8"))
self.display.vvv(f"Returned Data: {json.dumps(json_data, indent=4, sort_keys=True)}")
if 200 <= response.status <= 299:
if HTTPStatus(response.status).is_success:
return json_data
if response.status == 404 and json_data["status"] == "No objects found.":
if response.status == HTTPStatus.NOT_FOUND and json_data["status"] == "No objects found.":
raise AnsibleParserError(f"API returned no data -- Response: {response.status} - {json_data['status']}")
if response.status == 401:
if response.status == HTTPStatus.UNAUTHORIZED:
raise AnsibleParserError(
f"API was unable to complete query -- Response: {response.status} - {json_data['status']}"
)
if response.status == 500:
if response.status == HTTPStatus.INTERNAL_SERVER_ERROR:
raise AnsibleParserError(f"API Response - {json_data['status']} - {json_data['errors']}")
raise AnsibleParserError(f"Unexpected data returned - {json_data['status']} - {json_data['errors']}")

5
plugins/lookup/github_app_access_token.py
View file

@ -97,6 +97,7 @@ except ImportError:
import json
import time
from http import HTTPStatus
from urllib.error import HTTPError
from ansible.errors import AnsibleError, AnsibleOptionsError
@ -177,9 +178,9 @@ def post_request(generated_jwt, installation_id, api_base):
display.vvv(f"Error returned: {error_body}")
except Exception:
error_body = {}
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
raise AnsibleError("Github return error. Please confirm your installation_id value is valid") from e
elif e.code == 401:
elif e.code == HTTPStatus.UNAUTHORIZED:
raise AnsibleError("Github return error. Please confirm your private key is valid") from e
raise AnsibleError(f"Unexpected data returned: {e} -- {error_body}") from e
response_body = response.read()

11
plugins/module_utils/consul.py
View file

@ -8,6 +8,7 @@ import copy
import json
import re
import typing as t
from http import HTTPStatus
from urllib import error as urllib_error
from urllib.parse import urlencode
@ -41,7 +42,8 @@ class RequestError(Exception):
def handle_consul_response_error(response):
if 400 <= response.status_code < 600:
status = HTTPStatus(response.status_code)
if status.is_client_error or status.is_server_error:
raise RequestError(f"{response.status_code} {response.content}")
@ -242,9 +244,9 @@ class _ConsulModule:
try:
return self.get(url)
except RequestError as e:
if e.status == 404:
if e.status == HTTPStatus.NOT_FOUND:
return
elif e.status == 403 and b"ACL not found" in e.response_data:
elif e.status == HTTPStatus.FORBIDDEN and b"ACL not found" in e.response_data:
return
raise
@ -321,7 +323,8 @@ class _ConsulModule:
)
raise
if 400 <= status < 600:
_status = HTTPStatus(status)
if _status.is_client_error or _status.is_server_error:
raise RequestError(status, response_data)
if response_data:

9
plugins/module_utils/gandi_livedns_api.py
View file

@ -6,6 +6,7 @@ from __future__ import annotations
import json
import typing as t
from http import HTTPStatus
from ansible.module_utils.urls import fetch_url
@ -60,7 +61,7 @@ class GandiLiveDNSAPI:
resp, info = fetch_url(self.module, self.api_endpoint + api_call, headers=headers, data=data, method=method)
error_msg = ""
if info["status"] >= 400 and (info["status"] != 404 or error_on_404):
if info["status"] >= HTTPStatus.BAD_REQUEST and (info["status"] != HTTPStatus.NOT_FOUND or error_on_404):
err_s = self.error_strings.get(info["status"], "")
error_msg = f"API Error {err_s}: {self._build_error_message(self.module, info)}"
@ -112,7 +113,7 @@ class GandiLiveDNSAPI:
records, status = self._gandi_api_call(url, error_on_404=False)
if status == 404:
if status == HTTPStatus.NOT_FOUND:
return []
if not isinstance(records, list):
@ -135,8 +136,8 @@ class GandiLiveDNSAPI:
record, status = self._gandi_api_call(url, method="POST", payload=new_record)
if status in (
200,
201,
HTTPStatus.OK,
HTTPStatus.CREATED,
):
return new_record

5
plugins/module_utils/hwc_utils.py
View file

@ -8,6 +8,7 @@ import re
import time
import traceback
import typing as t
from http import HTTPStatus
THIRD_LIBRARIES_IMP_ERR = None
try:
@ -70,7 +71,7 @@ def session_method_wrapper(f):
raise HwcClientException(0, f"Parsing response to json failed, error: {ex}") from ex
code = r.status_code
if code not in [200, 201, 202, 203, 204, 205, 206, 207, 208, 226]:
if not HTTPStatus(code).is_success:
msg = ""
for i in ["message", "error.message"]:
try:
@ -81,7 +82,7 @@ def session_method_wrapper(f):
else:
msg = str(result)
if code == 404:
if code == HTTPStatus.NOT_FOUND:
raise HwcClientException404(msg)
raise HwcClientException(code, msg)

35
plugins/module_utils/identity/keycloak/keycloak.py
View file

@ -8,6 +8,7 @@ import copy
import json
import traceback
import typing as t
from http import HTTPStatus
from urllib.error import HTTPError
from urllib.parse import quote, urlencode
@ -418,7 +419,7 @@ class KeycloakAPI:
validate_certs=self.validate_certs,
)
except HTTPError as e:
if e.code != 401:
if e.code != HTTPStatus.UNAUTHORIZED:
raise e
return e
@ -440,7 +441,7 @@ class KeycloakAPI:
r = make_request_catching_401(headers)
except KeycloakError as e:
# Token refresh returns 400 if token is expired/invalid, so continue on if we get a 400
if e.authError is not None and e.authError.code != 400: # type: ignore # TODO!
if e.authError is not None and e.authError.code != HTTPStatus.BAD_REQUEST: # type: ignore # TODO!
raise e
if isinstance(r, Exception):
@ -465,7 +466,7 @@ class KeycloakAPI:
r = make_request_catching_401(headers)
except KeycloakError as e:
# Token refresh returns 400 if token is expired/invalid, so continue on if we get a 400
if e.authError is not None and e.authError.code != 400: # type: ignore # TODO!
if e.authError is not None and e.authError.code != HTTPStatus.BAD_REQUEST: # type: ignore # TODO!
raise e
if isinstance(r, Exception):
@ -496,7 +497,7 @@ class KeycloakAPI:
return self._request_and_deserialize(realm_info_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain realm {realm}: {e}", exception=traceback.format_exc())
@ -525,7 +526,7 @@ class KeycloakAPI:
return self._request_and_deserialize(realm_keys_metadata_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain realm {realm}: {e}", exception=traceback.format_exc())
@ -551,7 +552,7 @@ class KeycloakAPI:
return self._request_and_deserialize(realm_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain realm {realm}: {e}", exception=traceback.format_exc())
@ -718,7 +719,7 @@ class KeycloakAPI:
return self._request_and_deserialize(client_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain client {id} for realm {realm}: {e}")
@ -1310,7 +1311,7 @@ class KeycloakAPI:
return self._request_and_deserialize(clientscope_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch clientscope {cid} in realm {realm}: {e}")
@ -1434,7 +1435,7 @@ class KeycloakAPI:
return self._request_and_deserialize(protocolmapper_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch protocolmapper {pid} in realm {realm}: {e}")
@ -1641,7 +1642,7 @@ class KeycloakAPI:
return self._request_and_deserialize(clientsecret_url, method="POST")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain clientsecret of client {id} for realm {realm}: {e}")
@ -1661,7 +1662,7 @@ class KeycloakAPI:
return self._request_and_deserialize(clientsecret_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not obtain clientsecret of client {id} for realm {realm}: {e}")
@ -1695,7 +1696,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(groups_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch group {gid} in realm {realm}: {e}")
@ -1967,7 +1968,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(role_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch role {name} in realm {realm}: {e}")
@ -2169,7 +2170,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(role_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch role {name} in client {clientid} of realm {realm}: {e}")
@ -2597,7 +2598,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(idp_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch identity provider {alias} in realm {realm}: {e}")
@ -2682,7 +2683,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(mapper_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(
@ -2767,7 +2768,7 @@ class KeycloakAPI:
try:
return self._request_and_deserialize(comp_url, method="GET")
except HTTPError as e:
if e.code == 404:
if e.code == HTTPStatus.NOT_FOUND:
return None
else:
self.fail_request(e, msg=f"Could not fetch component {cid} in realm {realm}: {e}")

5
plugins/module_utils/ipa.py
View file

@ -17,6 +17,7 @@ import re
import socket
import typing as t
import uuid
from http import HTTPStatus
from urllib.parse import quote
from ansible.module_utils.basic import AnsibleFallbackNotFound, env_fallback
@ -90,7 +91,7 @@ class IPAClient:
module=self.module, url=url, data=to_bytes(data), headers=headers, timeout=self.timeout
)
status_code = info["status"]
if status_code not in [200, 201, 204]:
if status_code not in (HTTPStatus.OK, HTTPStatus.CREATED, HTTPStatus.NO_CONTENT):
self._fail("login", info["msg"])
self.headers = {"Cookie": info.get("set-cookie")}
@ -145,7 +146,7 @@ class IPAClient:
use_gssapi=self.use_gssapi,
)
status_code = info["status"]
if status_code not in [200, 201, 204]:
if status_code not in (HTTPStatus.OK, HTTPStatus.CREATED, HTTPStatus.NO_CONTENT):
self._fail(method, info["msg"])
except Exception as e:
self._fail(f"post {method}", f"{e}")

5
plugins/module_utils/memset.py
View file

@ -12,6 +12,7 @@
from __future__ import annotations
import urllib.error as urllib_error
from http import HTTPStatus
from urllib.parse import urlencode
from ansible.module_utils.basic import json
@ -93,7 +94,7 @@ def check_zone_domain(data, domain):
"""
exists = False
if data.status_code in [201, 200]:
if data.status_code in (HTTPStatus.CREATED, HTTPStatus.OK):
for zone_domain in data.json():
if zone_domain["domain"] == domain:
exists = True
@ -108,7 +109,7 @@ def check_zone(data, name):
counter = 0
exists = False
if data.status_code in [201, 200]:
if data.status_code in (HTTPStatus.CREATED, HTTPStatus.OK):
for zone in data.json():
if zone["nickname"] == name:
counter += 1

15
plugins/module_utils/net_tools/pritunl/api.py
View file

@ -15,6 +15,7 @@ import json
import time
import typing as t
import uuid
from http import HTTPStatus
from ansible.module_utils.urls import open_url
@ -158,7 +159,7 @@ def list_pritunl_organizations(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException("Could not retrieve organizations from Pritunl")
else:
for org in json.loads(response.read()):
@ -190,7 +191,7 @@ def list_pritunl_users(
organization_id=organization_id,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException("Could not retrieve users from Pritunl")
else:
for user in json.loads(response.read()):
@ -220,7 +221,7 @@ def post_pritunl_organization(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException(f"Could not add organization {organization_name} to Pritunl")
# The user PUT request returns the updated user object
return json.loads(response.read())
@ -246,7 +247,7 @@ def post_pritunl_user(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException(f"Could not remove user {user_id} from organization {organization_id} from Pritunl")
# user POST request returns an array of a single item,
# so return this item instead of the list
@ -262,7 +263,7 @@ def post_pritunl_user(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException(f"Could not update user {user_id} from organization {organization_id} from Pritunl")
# The user PUT request returns the updated user object
return json.loads(response.read())
@ -279,7 +280,7 @@ def delete_pritunl_organization(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException(f"Could not remove organization {organization_id} from Pritunl")
return json.loads(response.read())
@ -297,7 +298,7 @@ def delete_pritunl_user(
validate_certs=validate_certs,
)
if response.getcode() != 200:
if response.getcode() != HTTPStatus.OK:
raise PritunlException(f"Could not remove user {user_id} from organization {organization_id} from Pritunl")
return json.loads(response.read())

11
plugins/module_utils/ocapi_utils.py
View file

@ -9,6 +9,7 @@ import json
import os
import typing as t
import uuid
from http import HTTPStatus
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
@ -91,7 +92,7 @@ class OcapiUtils:
use_proxy=True,
timeout=self.timeout,
)
if resp.status != 204:
if resp.status != HTTPStatus.NO_CONTENT:
data = json.loads(resp.read())
else:
data = ""
@ -390,7 +391,7 @@ class OcapiUtils:
job_uri = self.get_uri_with_slot_number_query_param(job_uri)
response = self.get_request(job_uri)
if response["ret"] is False:
if response.get("status") == 404:
if response.get("status") == HTTPStatus.NOT_FOUND:
# Job not found -- assume 0%
return {
"ret": True,
@ -436,7 +437,7 @@ class OcapiUtils:
return {"ret": False, "changed": False, "msg": "Cannot delete job because it is in progress."}
if response["ret"] is False:
if response["status"] == 404:
if response["status"] == HTTPStatus.NOT_FOUND:
return {"ret": True, "changed": False, "msg": "Job already deleted."}
return response
if self.module.check_mode:
@ -445,9 +446,9 @@ class OcapiUtils:
# Do the DELETE (unless we are in check mode)
response = self.delete_request(job_uri, etag)
if response["ret"] is False:
if response["status"] == 404:
if response["status"] == HTTPStatus.NOT_FOUND:
return {"ret": True, "changed": False}
elif response["status"] == 409:
elif response["status"] == HTTPStatus.CONFLICT:
return {"ret": False, "changed": False, "msg": "Cannot delete job because it is in progress."}
return response
return {"ret": True, "changed": True}

8
plugins/module_utils/online.py
View file

@ -7,6 +7,7 @@ from __future__ import annotations
import json
import sys
import typing as t
from http import HTTPStatus
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.urls import fetch_url
@ -60,7 +61,12 @@ class Response:
@property
def ok(self):
return self.status_code in (200, 201, 202, 204)
return self.status_code in (
HTTPStatus.OK,
HTTPStatus.CREATED,
HTTPStatus.ACCEPTED,
HTTPStatus.NO_CONTENT,
)
class Online:

11
plugins/module_utils/oracle/oci_utils.py
View file

@ -18,6 +18,7 @@ import typing as t
# (TODO: remove next line!)
from datetime import datetime # noqa: F401, pylint: disable=unused-import
from http import HTTPStatus
from operator import eq
try:
@ -1381,7 +1382,7 @@ def delete_and_wait(
except MaximumWaitTimeExceeded as ex:
module.fail_json(msg=str(ex))
except ServiceError as ex:
if ex.status != 404:
if ex.status != HTTPStatus.NOT_FOUND:
module.fail_json(msg=ex.message)
else:
# While waiting for resource to get into terminated state, if the resource is not found.
@ -1405,9 +1406,9 @@ def delete_and_wait(
# DNS API throws a 400 InvalidParameter when a zone id is provided for zone_name_or_id and if the zone
# resource is not available, instead of the expected 404. So working around this for now.
if isinstance(client, oci.dns.DnsClient):
if ex.status == 400 and ex.code == "InvalidParameter":
if ex.status == HTTPStatus.BAD_REQUEST and ex.code == "InvalidParameter":
_debug(f"Resource {resource_type} with {kwargs_get} already deleted. So returning changed=False")
elif ex.status != 404:
elif ex.status != HTTPStatus.NOT_FOUND:
module.fail_json(msg=ex.message)
result[resource_type] = dict()
return result
@ -1654,7 +1655,7 @@ def get_existing_resource(target_fn, module: AnsibleModule, **kwargs):
response = call_with_backoff(target_fn, **kwargs)
existing_resource = response.data
except ServiceError as ex:
if ex.status != 404:
if ex.status != HTTPStatus.NOT_FOUND:
module.fail_json(msg=ex.message)
return existing_resource
@ -1682,7 +1683,7 @@ def get_attached_instance_info(
# Pass ServiceError due to authorization issue in accessing volume attachments of a compartment
except ServiceError as ex:
if ex.status == 404:
if ex.status == HTTPStatus.NOT_FOUND:
pass
else:

24
plugins/module_utils/redfish_utils.py
View file

@ -11,6 +11,7 @@ import random
import string
import time
import typing as t
from http import HTTPStatus
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
@ -442,7 +443,8 @@ class RedfishUtils:
"""
msg = http_client.responses.get(error.code, "")
data = None
if error.code >= 400:
code = HTTPStatus(error.code)
if code.is_client_error or code.is_server_error:
try:
body = error.read().decode("utf-8")
data = json.loads(body)
@ -1498,8 +1500,8 @@ class RedfishUtils:
response = self.post_request(self.root_uri + self.accounts_uri, payload)
if not response["ret"]:
if response.get("status") == 405:
# if POST returned a 405, try to add via PATCH
if response.get("status") == HTTPStatus.METHOD_NOT_ALLOWED:
# if POST is not allowed, try to add via PATCH
return self.add_user_via_patch(user)
else:
return response
@ -1549,8 +1551,8 @@ class RedfishUtils:
response = self.delete_request(self.root_uri + uri)
if not response["ret"]:
if response.get("status") == 405:
# if DELETE returned a 405, try to delete via PATCH
if response.get("status") == HTTPStatus.METHOD_NOT_ALLOWED:
# if DELETE is not allowed, try to delete via PATCH
return self.delete_user_via_patch(user, uri=uri, data=data)
else:
return response
@ -1868,7 +1870,7 @@ class RedfishUtils:
operation_results = {"status": None, "messages": [], "handle": None, "ret": True, "resets_requested": []}
if response.status == 204:
if response.status == HTTPStatus.NO_CONTENT:
# No content; successful, but nothing to return
# Use the Redfish "Completed" enum from TaskState for the operation status
operation_results["status"] = "Completed"
@ -1877,7 +1879,7 @@ class RedfishUtils:
# Determine the next handle, if any
operation_results["handle"] = handle
if response.status == 202:
if response.status == HTTPStatus.ACCEPTED:
# Task generated; get the task monitor URI
operation_results["handle"] = response.getheader("Location", handle)
@ -1891,15 +1893,17 @@ class RedfishUtils:
else:
# Error response body, which is a bit of a misnomer since it is used in successful action responses
operation_results["status"] = "Completed"
if response.status >= 400:
_status = HTTPStatus(response.status)
if _status.is_client_error or _status.is_server_error:
operation_results["status"] = "Exception"
operation_results["messages"] = data.get("error", {}).get("@Message.ExtendedInfo", [])
else:
# No response body (or malformed); build based on status code
operation_results["status"] = "Completed"
if response.status == 202:
_status = HTTPStatus(response.status)
if _status == HTTPStatus.ACCEPTED:
operation_results["status"] = "New"
elif response.status >= 400:
elif _status.is_client_error or _status.is_server_error:
operation_results["status"] = "Exception"
# Clear out the handle if the operation is complete

11
plugins/module_utils/rundeck.py
View file

@ -7,6 +7,7 @@ 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
@ -74,13 +75,15 @@ def api_request(
},
)
if info["status"] == 403:
_status = HTTPStatus(info["status"])
if _status == HTTPStatus.FORBIDDEN:
module.fail_json(msg="Token authorization failed", execution_info=json.loads(info["body"]))
elif info["status"] == 404:
elif _status == HTTPStatus.NOT_FOUND:
return None, info
elif info["status"] == 409:
elif _status == HTTPStatus.CONFLICT:
module.fail_json(msg="Job executions limit reached", execution_info=json.loads(info["body"]))
elif info["status"] >= 500:
elif _status.is_server_error:
module.fail_json(msg="Rundeck API error", execution_info=json.loads(info["body"]))
try:

10
plugins/module_utils/scaleway.py
View file

@ -12,6 +12,7 @@ import sys
import time
import traceback
import typing as t
from http import HTTPStatus
from urllib.parse import urlencode
from ansible.module_utils.basic import env_fallback, missing_required_lib
@ -199,7 +200,12 @@ class Response:
@property
def ok(self):
return self.status_code in (200, 201, 202, 204)
return self.status_code in (
HTTPStatus.OK,
HTTPStatus.CREATED,
HTTPStatus.ACCEPTED,
HTTPStatus.NO_CONTENT,
)
class Scaleway:
@ -302,7 +308,7 @@ class Scaleway:
self.module.debug(f"fetch_state of resource: {resource['id']}")
response = self.get(path=f"{self.api_path}/{resource['id']}")
if response.status_code == 404:
if response.status_code == HTTPStatus.NOT_FOUND:
return "absent"
if not response.ok:

3
plugins/module_utils/source_control/bitbucket.py
View file

@ -6,6 +6,7 @@ from __future__ import annotations
import json
import typing as t
from http import HTTPStatus
from ansible.module_utils.basic import env_fallback
from ansible.module_utils.urls import basic_auth_header, fetch_url
@ -56,7 +57,7 @@ class BitbucketHelper:
headers=headers,
)
if info["status"] == 200:
if info["status"] == HTTPStatus.OK:
self.access_token = content["access_token"]
else:
self.module.fail_json(msg=f"Failed to retrieve access token: {info}")

16
plugins/module_utils/utm_utils.py
View file

@ -13,6 +13,7 @@ from __future__ import annotations
import json
import typing as t
from http import HTTPStatus
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.urls import fetch_url
@ -116,7 +117,8 @@ class UTM:
returns the info for an object in utm
"""
info, result = self._lookup_entry(self.module, self.request_url)
if info["status"] >= 400:
_status = HTTPStatus(info["status"])
if _status.is_client_error or _status.is_server_error:
self.module.fail_json(result=json.loads(info))
else:
if result is None:
@ -133,7 +135,8 @@ class UTM:
is_changed = False
info, result = self._lookup_entry(self.module, self.request_url)
if info["status"] >= 400:
_status = HTTPStatus(info["status"])
if _status.is_client_error or _status.is_server_error:
self.module.fail_json(result=json.loads(info))
else:
data_as_json_string = self.module.jsonify(self.module.params)
@ -141,7 +144,8 @@ class UTM:
response, info = fetch_url(
self.module, self.request_url, method="POST", headers=combined_headers, data=data_as_json_string
)
if info["status"] >= 400:
_status = HTTPStatus(info["status"])
if _status.is_client_error or _status.is_server_error:
self.module.fail_json(msg=json.loads(info["body"]))
is_changed = True
result = self._clean_result(json.loads(response.read()))
@ -154,7 +158,8 @@ class UTM:
headers=combined_headers,
data=data_as_json_string,
)
if info["status"] >= 400:
_status = HTTPStatus(info["status"])
if _status.is_client_error or _status.is_server_error:
self.module.fail_json(msg=json.loads(info["body"]))
is_changed = True
result = self._clean_result(json.loads(response.read()))
@ -187,7 +192,8 @@ class UTM:
headers={"Accept": "application/json", "X-Restd-Err-Ack": "all"},
data=self.module.jsonify(self.module.params),
)
if info["status"] >= 400:
_status = HTTPStatus(info["status"])
if _status.is_client_error or _status.is_server_error:
self.module.fail_json(msg=json.loads(info["body"]))
else:
is_changed = True