Coverage for gws-app/gws/base/ows/client/request.py: 41%

1"""HTTP requests to OWS services.""" 

2 

3import gws 

4import gws.lib.net 

5 

6 

7_ows_error_strings = '<ServiceException', '<ServerException', '<ows:ExceptionReport' 

8 

9 

10class Args(gws.Data): 

11 """Arguments for an OWS request.""" 

12 

13 method: gws.RequestMethod 

14 """Request method, ``GET`` by default.""" 

15 headers: dict 

16 """HTTP headers.""" 

17 params: dict 

18 """Additional request parameters.""" 

19 protocol: gws.OwsProtocol 

20 """Service protocol, sent as ``SERVICE``.""" 

21 url: str 

22 """Service URL.""" 

23 verb: gws.OwsVerb 

24 """Request type, sent as ``REQUEST``.""" 

25 version: str 

26 """Service version, sent as ``VERSION`` if not empty.""" 

27 

28 

29def get_url(url: str, **kwargs) -> gws.lib.net.HTTPResponse: 

30 """Send an HTTP request to an OWS service. 

31 

32 The response is checked for an OWS exception document, regardless of the 

33 HTTP status, since some servers return errors with status 200. 

34 

35 Args: 

36 url: Request URL. 

37 **kwargs: Options for ``gws.lib.net.http_request``. 

38 

39 Returns: 

40 The HTTP response. 

41 

42 Raises: 

43 ``gws.ExternalServiceError``: If the service returns an exception document or the request fails. 

44 """ 

45 

46 res = gws.lib.net.http_request(url, **kwargs) 

47 

48 # some folks serve OWS error documents with the status 200 

49 # therefore, we check for an ows error message, no matter what the status code says 

50 # we can get big image responses here, so be careful and don't blindly decode everything 

51 

52 if res.content.startswith(b'<') or 'xml' in res.content_type: 

53 text = str(res.content[:1024], encoding='utf8', errors='ignore') 

54 text_lower = text.lower() 

55 for err_string in _ows_error_strings: 

56 if err_string.lower() in text_lower: 

57 raise gws.ExternalServiceError(text) 

58 try: 

59 res.raise_if_failed() 

60 except gws.lib.net.HTTPError as exc: 

61 raise gws.ExternalServiceError(f'network error: {url!r} -> {exc}') from exc 

62 return res 

63 

64 

65def get(args: Args, **kwargs) -> gws.lib.net.HTTPResponse: 

66 """Send an OWS request and return the raw response. 

67 

68 Args: 

69 args: Request arguments. 

70 **kwargs: Options for ``gws.lib.net.http_request``. 

71 

72 Returns: 

73 The HTTP response. 

74 

75 Raises: 

76 ``gws.ExternalServiceError``: If the service returns an exception document or the request fails. 

77 """ 

78 

79 params = { 

80 'SERVICE': str(args.protocol).upper(), 

81 'REQUEST': args.verb, 

82 } 

83 if args.version: 

84 params['VERSION'] = args.version 

85 if args.params: 

86 params.update(gws.u.to_upper_dict(args.params)) 

87 

88 return get_url(args.url, method=args.method or gws.RequestMethod.GET, params=params, headers=args.headers, **kwargs) 

89 

90 

91def get_text(args: Args, **kwargs) -> str: 

92 """Send an OWS request and return the response text. 

93 

94 Args: 

95 args: Request arguments. 

96 **kwargs: Options for ``gws.lib.net.http_request``. 

97 

98 Returns: 

99 The response text. 

100 

101 Raises: 

102 ``gws.ExternalServiceError``: If the service returns an exception document or the request fails. 

103 """ 

104 

105 res = get(args, **kwargs) 

106 return res.text