Coverage for langsmith/_internal/_patch.py: 0%

1import functools 

2 

3from urllib3 import __version__ as urllib3version # type: ignore[import-untyped] 

4from urllib3 import connection # type: ignore[import-untyped] 

5 

6 

7def _ensure_str(s, encoding="utf-8", errors="strict") -> str: 

8 if isinstance(s, str): 

9 return s 

10 

11 if isinstance(s, bytes): 

12 return s.decode(encoding, errors) 

13 return str(s) 

14 

15 

16# Copied from https://github.com/urllib3/urllib3/blob/1c994dfc8c5d5ecaee8ed3eb585d4785f5febf6e/src/urllib3/connection.py#L231 

17def request(self, method, url, body=None, headers=None): 

18 """Make the request. 

19 

20 This function is based on the urllib3 request method, with modifications 

21 to handle potential issues when using vcrpy in concurrent workloads. 

22 

23 Args: 

24 self: The HTTPConnection instance. 

25 method (str): The HTTP method (e.g., 'GET', 'POST'). 

26 url (str): The URL for the request. 

27 body (Optional[Any]): The body of the request. 

28 headers (Optional[dict]): Headers to send with the request. 

29 

30 Returns: 

31 The result of calling the parent request method. 

32 """ 

33 # Update the inner socket's timeout value to send the request. 

34 # This only triggers if the connection is re-used. 

35 if getattr(self, "sock", None) is not None: 

36 self.sock.settimeout(self.timeout) 

37 

38 if headers is None: 

39 headers = {} 

40 else: 

41 # Avoid modifying the headers passed into .request() 

42 headers = headers.copy() 

43 if "user-agent" not in (_ensure_str(k.lower()) for k in headers): 

44 headers["User-Agent"] = connection._get_default_user_agent() 

45 # The above is all the same ^^^ 

46 # The following is different: 

47 return self._parent_request(method, url, body=body, headers=headers) 

48 

49 

50_PATCHED = False 

51 

52 

53def patch_urllib3(): 

54 """Patch the request method of urllib3 to avoid type errors when using vcrpy. 

55 

56 In concurrent workloads (such as the tracing background queue), the 

57 connection pool can get in a state where an HTTPConnection is created 

58 before vcrpy patches the HTTPConnection class. In urllib3 >= 2.0 this isn't 

59 a problem since they use the proper super().request(...) syntax, but in older 

60 versions, super(HTTPConnection, self).request is used, resulting in a TypeError 

61 since self is no longer a subclass of "HTTPConnection" (which at this point 

62 is vcr.stubs.VCRConnection). 

63 

64 This method patches the class to fix the super() syntax to avoid mixed inheritance. 

65 In the case of the LangSmith tracing logic, it doesn't really matter since we always 

66 exclude cache checks for calls to LangSmith. 

67 

68 The patch is only applied for urllib3 versions older than 2.0. 

69 """ 

70 global _PATCHED 

71 if _PATCHED: 

72 return 

73 from packaging import version 

74 

75 if version.parse(urllib3version) >= version.parse("2.0"): 

76 _PATCHED = True 

77 return 

78 

79 # Lookup the parent class and its request method 

80 parent_class = connection.HTTPConnection.__bases__[0] 

81 parent_request = parent_class.request 

82 

83 def new_request(self, *args, **kwargs): 

84 """Handle parent request. 

85 

86 This method binds the parent's request method to self and then 

87 calls our modified request function. 

88 """ 

89 self._parent_request = functools.partial(parent_request, self) 

90 return request(self, *args, **kwargs) 

91 

92 connection.HTTPConnection.request = new_request 

93 _PATCHED = True