Coverage for src/prisma/_sync_http.py: 76%

1from typing import Any 

2from typing_extensions import override 

3 

4import httpx 

5 

6from ._types import Method 

7from .http_abstract import AbstractHTTP, AbstractResponse 

8 

9__all__ = ('HTTP', 'SyncHTTP', 'Response', 'client') 

10 

11 

12class SyncHTTP(AbstractHTTP[httpx.Client, httpx.Response]): 

13 session: httpx.Client 

14 

15 @override 

16 def download(self, url: str, dest: str) -> None: 

17 with self.session.stream('GET', url, timeout=None) as resp: 

18 resp.raise_for_status() 

19 with open(dest, 'wb') as fd: 

20 for chunk in resp.iter_bytes(): 

21 fd.write(chunk) 

22 

23 @override 

24 def request(self, method: Method, url: str, **kwargs: Any) -> 'Response': 

25 return Response(self.session.request(method, url, **kwargs)) 

26 

27 @override 

28 def open(self) -> None: 

29 self.session = httpx.Client(**self.session_kwargs) 

30 

31 @override 

32 def close(self) -> None: 

33 if self.should_close(): 

34 self.session.close() 

35 self.session = None # type: ignore[assignment] 

36 

37 def __del__(self) -> None: 

38 self.close() 

39 

40 

41HTTP = SyncHTTP 

42 

43client: HTTP = HTTP() 

44 

45 

46class Response(AbstractResponse[httpx.Response]): 

47 __slots__ = () 

48 

49 @property 

50 @override 

51 def status(self) -> int: 

52 return self.original.status_code 

53 

54 @property 

55 @override 

56 def headers(self) -> httpx.Headers: 

57 return self.original.headers 

58 

59 @override 

60 def json(self, **kwargs: Any) -> Any: 

61 return self.original.json(**kwargs) 

62 

63 @override 

64 def text(self, **kwargs: Any) -> str: 

65 return self.original.content.decode(**kwargs)