Skip to content

operations

get_req_client(headers=default_headers, timeout=None)

Return a customized HTTPX client.

Source code in src/red_utils/ext/httpx_utils/operations.py 
51
52
53
54
55
56
57
58
59
def get_req_client(
    headers: dict | None = default_headers, timeout: int | None = None
) -> httpx.Client:
    """Return a customized HTTPX client."""
    _client: Client = Client(headers=headers, timeout=timeout)

    validate_client(_client)

    return _client

make_request(client=None, url=None, method='GET', headers=None, timeout=None, data=None)

Make a request with HTTPX.

Source code in src/red_utils/ext/httpx_utils/operations.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def make_request(
    client: Client = None,
    url: str = None,
    method: str = "GET",
    headers: dict = None,
    timeout: int | None = None,
    data: dict | None = None,
) -> httpx.Response:
    """Make a request with HTTPX."""
    validate_method(method)

    if not client:
        print(
            f"[HTTPX] [WARNING] No request client passed to make_request() function. Getting default client."
        )

        client = get_req_client(headers=headers, timeout=timeout)

    validate_client(client=client)

    if not isinstance(client, Client):
        raise TypeError(
            f"Invalid type for client: {type(client)}. Must be of type {type(Client)}"
        )

    if not url:
        raise ValueError("Missing request URL")

    if not isinstance(url, str):
        raise TypeError(f"Invalid type for request URL: {type(url)}")

    if timeout:
        if not isinstance(timeout, int):
            raise TypeError(f"Invalid type for timeout: {type(timeout)}")

    try:
        ## Determine type of request to make
        match method:
            case "GET":
                client = get_req_client(headers=headers, timeout=timeout)

                res = client.get(url=url)

            case _:
                raise ValueError(f"Invalid method: {method}")

        return res

    except Exception as exc:
        raise Exception(f"Unhandled exception creating request client. Details: {exc}")

merge_headers(original_headers=default_headers, update_vals=None)

Merge header dicts into new headers dict.

Source code in src/red_utils/ext/httpx_utils/operations.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def merge_headers(
    original_headers: dict[str, str] = default_headers,
    update_vals: dict[str, str] = None,
) -> dict[str, str]:
    """Merge header dicts into new headers dict."""
    validate_headers(original_headers)
    validate_headers(update_vals)

    try:
        _headers: dict = {**update_vals, **original_headers}

    except Exception as exc:
        raise Exception(f"Unhandled exception merging header dicts. Details: {exc}")

    return _headers