(err_ctx, # type: HTTPErrorContext
mapping=None, # type: Dict[str, CouchbaseException]
err_info=None # type: Dict[str, Any]
)
| 2028 | |
| 2029 | @staticmethod |
| 2030 | def _parse_http_context(err_ctx, # type: HTTPErrorContext |
| 2031 | mapping=None, # type: Dict[str, CouchbaseException] |
| 2032 | err_info=None # type: Dict[str, Any] |
| 2033 | ) -> Optional[CouchbaseException]: |
| 2034 | from couchbase._utils import is_null_or_empty |
| 2035 | |
| 2036 | compiled_map = {} |
| 2037 | if mapping: |
| 2038 | compiled_map = {{str: re.compile}.get( |
| 2039 | type(k), lambda x: x)(k): v for k, v in mapping.items()} |
| 2040 | |
| 2041 | exc_msg = err_info.get('error_message', None) if err_info else None |
| 2042 | if not is_null_or_empty(exc_msg): |
| 2043 | exc_class = ErrorMapper._process_mapping(compiled_map, exc_msg) |
| 2044 | if exc_class is not None: |
| 2045 | return exc_class |
| 2046 | |
| 2047 | if not is_null_or_empty(err_ctx.response_body): |
| 2048 | err_text = err_ctx.response_body |
| 2049 | exc_class = ErrorMapper._process_mapping(compiled_map, err_text) |
| 2050 | if exc_class is not None: |
| 2051 | return exc_class |
| 2052 | |
| 2053 | exc_class = ErrorMapper._parse_http_response_body(compiled_map, err_text) |
| 2054 | if exc_class is not None: |
| 2055 | return exc_class |
| 2056 | |
| 2057 | return None |
| 2058 | |
| 2059 | @staticmethod |
| 2060 | def _parse_kv_context(err_ctx, # type: KeyValueErrorContext |
no test coverage detected