Coverage for haystack/utils/http_client.py: 100%

1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai> 

2# 

3# SPDX-License-Identifier: Apache-2.0 

4 

5from typing import Any, Literal, overload 

6 

7import httpx 

8 

9 

10@overload 

11def init_http_client( 

12 http_client_kwargs: dict[str, Any] | None = ..., async_client: Literal[False] = ... 

13) -> httpx.Client | None: ... 

14@overload 

15def init_http_client( 

16 http_client_kwargs: dict[str, Any] | None = ..., async_client: Literal[True] = ... 

17) -> httpx.AsyncClient | None: ... 

18def init_http_client( 

19 http_client_kwargs: dict[str, Any] | None = None, async_client: bool = False 

20) -> httpx.Client | httpx.AsyncClient | None: 

21 """ 

22 Initialize an httpx client based on the http_client_kwargs. 

23 

24 :param http_client_kwargs: 

25 The kwargs to pass to the httpx client. 

26 :param async_client: 

27 Whether to initialize an async client. 

28 

29 :returns: 

30 A httpx client or an async httpx client. 

31 """ 

32 if not http_client_kwargs: 

33 return None 

34 if not isinstance(http_client_kwargs, dict): 

35 raise TypeError("The parameter 'http_client_kwargs' must be a dictionary.") 

36 

37 # Create a copy to avoid modifying the original dict 

38 processed_kwargs = http_client_kwargs.copy() 

39 

40 # Handle limits parameter - convert dict to httpx.Limits object if needed 

41 if "limits" in processed_kwargs and isinstance(processed_kwargs["limits"], dict): 

42 limits_dict = processed_kwargs["limits"] 

43 processed_kwargs["limits"] = httpx.Limits(**limits_dict) 

44 

45 if async_client: 

46 return httpx.AsyncClient(**processed_kwargs) 

47 return httpx.Client(**processed_kwargs)