update requests for DRY

This commit is contained in:
Patrick Nisble 2026-01-12 08:37:16 +01:00
parent 2dca8c833c
commit 93a39ad889
1 changed files with 48 additions and 24 deletions

View File

@ -1,12 +1,18 @@
from json import dumps
from ssl import SSLContext, create_default_context from ssl import SSLContext, create_default_context
from urllib.error import HTTPError
from urllib.request import Request, urlopen from urllib.request import Request, urlopen
from urllib.parse import urlencode
def get( def request(
url: str, url: str,
method: str,
query_params: dict[str, str] | None = None, query_params: dict[str, str] | None = None,
data: dict[str, str] | None = None,
headers: dict[str, str] | None = None, headers: dict[str, str] | None = None,
) -> str: ) -> str:
try:
with urlopen( with urlopen(
Request( Request(
url url
@ -15,12 +21,32 @@ def get(
if query_params if query_params
else "" else ""
), ),
dumps(data).encode() if data else None,
headers=headers or {}, headers=headers or {},
method="GET", method=method,
), ),
context=create_default_context(), context=create_default_context(),
) as resp: ) as resp:
return resp.read().decode("utf-8") return resp.read().decode("utf-8")
except HTTPError as e:
e.add_note(e.url)
raise
def get(
url: str,
query_params: dict[str, str] | None = None,
headers: dict[str, str] | None = None,
) -> str:
return request(url, "GET", query_params=query_params, headers=headers)
def post(
url: str,
data: dict[str, str] | None = None,
headers: dict[str, str] | None = None,
) -> str:
return request(url, "POST", data=data, headers=headers)
def put( def put(
@ -28,13 +54,11 @@ def put(
data: dict[str, str] | None = None, data: dict[str, str] | None = None,
headers: dict[str, str] | None = None, headers: dict[str, str] | None = None,
) -> str: ) -> str:
with urlopen( return request(url, "PUT", data=data, headers=headers)
Request(
url,
str(data).encode(), def delete(
headers=headers or {}, url: str,
method="PUT", headers: dict[str, str] | None = None,
), ) -> str:
context=create_default_context(), return request(url, "DELETE", headers=headers)
) as resp:
return resp.read().decode("utf-8")