feat: Implement update_url_params

pull/84290/head
Hubert 2 weeks ago
parent 9d249432c4
commit c1f1cb33c2

@ -0,0 +1,2 @@
minor_changes:
- Add new utils function update_url_params to update url query parameters

@ -50,7 +50,7 @@ import urllib.error
import urllib.request
from contextlib import contextmanager
from http import cookiejar
from urllib.parse import unquote, urlparse, urlunparse
from urllib.parse import unquote, urlparse, urlunparse, ParseResult, parse_qs, urlencode
from urllib.request import BaseHandler
try:
@ -1102,6 +1102,44 @@ def prepare_multipart(fields):
)
def update_url_params(url, params):
"""Updates the query parameters of a given URL with additional parameters.
:arg url: str or ParseResult
The base URL or parsed URL object to update.
:arg params: dict
A dictionary of query parameters to add or update in the URL.
:returns: str
The updated URL with the new or modified query parameters.
This function parses the original URL, merges the given parameters with
existing ones, and returns a new URL string with the updated query parameters.
Example:
url = "http://example.com/path?existing_param=value"
params = {"new_param": "new_value", "existing_param": "updated_value"}
result = update_url_params(url, params)
// result: "http://example.com/path?existing_param=updated_value&new_param=new_value"
"""
if isinstance(url, ParseResult):
o = url
else:
o = urlparse(url)
_params = {
**parse_qs(o.query),
**params
}
new = o._replace(
query=urlencode(
_params,
doseq=True
)
)
return new.geturl()
#
# Module-related functions
#

@ -13,6 +13,7 @@ from ansible.module_utils.urls import (Request, open_url, cookiejar,
UnixHTTPHandler, UnixHTTPSConnection)
from ansible.module_utils.urls import HTTPRedirectHandler
from ansible.module_utils.urls import update_url_params
import pytest
from unittest.mock import call
@ -445,3 +446,15 @@ def test_open_url(urlopen_mock, install_opener_mock, mocker):
client_cert=None, client_key=None, cookies=None, use_gssapi=False,
unix_socket=None, ca_path=None, unredirected_headers=None, decompress=True,
ciphers=None, use_netrc=True)
def test_params_parameter():
payload = {'key1': 'value1', 'key2': 'value2'}
r = update_url_params('https://httpbin.org/get', payload)
assert r == "https://httpbin.org/get?key1=value1&key2=value2"
r = update_url_params('https://httpbin.org/get?key1=oldValue', payload)
assert r == "https://httpbin.org/get?key1=value1&key2=value2"
r = update_url_params('https://httpbin.org/get?key2=oldValue', payload)
assert r == "https://httpbin.org/get?key2=value2&key1=value1"
Loading…
Cancel
Save