(err_ctx, # type: KeyValueErrorContext
mapping, # type: Dict[str, CouchbaseException]
err_content=None # type: str
)
| 2058 | |
| 2059 | @staticmethod |
| 2060 | def _parse_kv_context(err_ctx, # type: KeyValueErrorContext |
| 2061 | mapping, # type: Dict[str, CouchbaseException] |
| 2062 | err_content=None # type: str |
| 2063 | ) -> Optional[CouchbaseException]: |
| 2064 | from couchbase._utils import is_null_or_empty |
| 2065 | |
| 2066 | compiled_map = {{str: re.compile}.get( |
| 2067 | type(k), lambda x: x)(k): v for k, v in mapping.items()} |
| 2068 | |
| 2069 | if not is_null_or_empty(err_content): |
| 2070 | exc_class = ErrorMapper._process_mapping(compiled_map, err_content) |
| 2071 | if exc_class is not None: |
| 2072 | return exc_class |
| 2073 | |
| 2074 | if err_ctx.retry_reasons is not None: |
| 2075 | for rr in err_ctx.retry_reasons: |
| 2076 | exc_class = ErrorMapper._process_mapping(compiled_map, rr) |
| 2077 | if exc_class is not None: |
| 2078 | return exc_class |
| 2079 | |
| 2080 | return None |
| 2081 | |
| 2082 | @classmethod |
| 2083 | def build_exception(cls, |
no test coverage detected