def handle_operation()

in firebase_admin/ml.py [0:0]

• 29 lines of code
• 7 McCabe index (conditional complexity)

    def handle_operation(self, operation, wait_for_operation=False, max_time_seconds=None):
        """Handles long running operations.

        Args:
            operation: The operation to handle.
            wait_for_operation: Should we allow polling for the operation to complete.
                If no polling is requested, a locked model will be returned instead.
            max_time_seconds: The maximum seconds to try polling for operation complete.
                (None for no limit)

        Returns:
            dict: A dictionary of the returned model properties.

        Raises:
            TypeError: if the operation is not a dictionary.
            ValueError: If the operation is malformed.
            UnknownError: If the server responds with an unexpected response.
            err: If the operation exceeds polling attempts or stop_time
        """
        if not isinstance(operation, dict):
            raise TypeError('Operation must be a dictionary.')

        if operation.get('done'):
            # Operations which are immediately done don't have an operation name
            if operation.get('response'):
                return operation.get('response')
            if operation.get('error'):
                raise _utils.handle_operation_error(operation.get('error'))
            raise exceptions.UnknownError(message='Internal Error: Malformed Operation.')

        op_name = _validate_operation_name(operation.get('name'))
        metadata = operation.get('metadata', {})
        metadata_type = metadata.get('@type', '')
        if not metadata_type.endswith('ModelOperationMetadata'):
            raise TypeError('Unknown type of operation metadata.')
        _, model_id = _validate_and_parse_name(metadata.get('name'))
        current_attempt = 0
        start_time = datetime.datetime.now()
        stop_time = (None if max_time_seconds is None else
                     start_time + datetime.timedelta(seconds=max_time_seconds))
        while wait_for_operation and not operation.get('done'):
            # We just got this operation. Wait before getting another
            # so we don't exceed the GetOperation maximum request rate.
            self._exponential_backoff(current_attempt, stop_time)
            operation = self.get_operation(op_name)
            current_attempt += 1

        if operation.get('done'):
            if operation.get('response'):
                return operation.get('response')
            if operation.get('error'):
                raise _utils.handle_operation_error(operation.get('error'))

        # If the operation is not complete or timed out, return a (locked) model instead
        return get_model(model_id).as_dict()