Coverage for src/prisma/http_abstract.py: 97%

1from abc import ABC, abstractmethod 

2from typing import ( 

3 Any, 

4 Dict, 

5 Type, 

6 Union, 

7 Generic, 

8 TypeVar, 

9 Optional, 

10 Coroutine, 

11 cast, 

12) 

13from typing_extensions import override 

14 

15from httpx import Limits, Headers, Timeout 

16 

17from .utils import _NoneType 

18from ._types import Method 

19from .errors import HTTPClientClosedError 

20 

21Session = TypeVar('Session') 

22Response = TypeVar('Response') 

23ReturnType = TypeVar('ReturnType') 

24MaybeCoroutine = Union[Coroutine[Any, Any, ReturnType], ReturnType] 

25 

26DEFAULT_CONFIG: Dict[str, Any] = { 

27 'limits': Limits(max_connections=1000), 

28 'timeout': Timeout(30), 

29} 

30 

31 

32class AbstractHTTP(ABC, Generic[Session, Response]): 

33 session_kwargs: Dict[str, Any] 

34 

35 __slots__ = ( 

36 '_session', 

37 'session_kwargs', 

38 ) 

39 

40 # NOTE: ParamSpec wouldn't be valid here: 

41 # https://github.com/microsoft/pyright/issues/2667 

42 def __init__(self, **kwargs: Any) -> None: 

43 # NoneType = not used yet 

44 # None = closed 

45 # Session = open 

46 self._session: Optional[Union[Session, Type[_NoneType]]] = _NoneType 

47 self.session_kwargs = { 

48 **DEFAULT_CONFIG, 

49 **kwargs, 

50 } 

51 

52 @abstractmethod 

53 def download(self, url: str, dest: str) -> MaybeCoroutine[None]: ... 

54 

55 @abstractmethod 

56 def request(self, method: Method, url: str, **kwargs: Any) -> MaybeCoroutine['AbstractResponse[Response]']: ... 

57 

58 @abstractmethod 

59 def open(self) -> None: ... 

60 

61 @abstractmethod 

62 def close(self) -> MaybeCoroutine[None]: ... 

63 

64 @property 

65 def closed(self) -> bool: 

66 return self._session is None 

67 

68 @property 

69 def session(self) -> Session: 

70 session = self._session 

71 if session is _NoneType: 

72 self.open() 

73 return cast(Session, self._session) 

74 

75 # TODO: make this not strict, just open a new session 

76 if session is None: 

77 raise HTTPClientClosedError() 

78 

79 return cast(Session, session) 

80 

81 @session.setter 

82 def session(self, value: Optional[Session]) -> None: # pyright: ignore[reportPropertyTypeMismatch] 

83 self._session = value 

84 

85 def should_close(self) -> bool: 

86 return self._session is not _NoneType and not self.closed 

87 

88 @override 

89 def __repr__(self) -> str: 

90 return str(self) 

91 

92 @override 

93 def __str__(self) -> str: 

94 return f'<HTTP closed={self.closed}>' 

95 

96 

97class AbstractResponse(ABC, Generic[Response]): 

98 original: Response 

99 

100 __slots__ = ('original',) 

101 

102 def __init__(self, original: Response) -> None: 

103 self.original = original 

104 

105 @property 

106 @abstractmethod 

107 def status(self) -> int: ... 

108 

109 @property 

110 @abstractmethod 

111 def headers(self) -> Headers: ... 

112 

113 @abstractmethod 

114 def json(self) -> MaybeCoroutine[Any]: ... 

115 

116 @abstractmethod 

117 def text(self) -> MaybeCoroutine[str]: ... 

118 

119 @override 

120 def __repr__(self) -> str: 

121 return str(self) 

122 

123 @override 

124 def __str__(self) -> str: 

125 return f'<Response wrapped={self.original} >'