How to use the dbt.exceptions.InternalException function in dbt

def collect_task_id(
        self, result: GCResult, task_id: TaskID
    ) -> None:
        """To collect a task ID, we just delete it from the tasks dict.

        You must hold the lock, as this mutates `tasks`.
        """
        try:
            state = self._remove_task_if_finished(task_id)
        except KeyError:
            # someone was mutating tasks while we had the lock, that's
            # not right!
            raise dbt.exceptions.InternalException(
                'Got a KeyError for task uuid={} during gc'
                .format(task_id)
            )

        return result.add_result(task_id=task_id, state=state)

def handle_singlethreaded(
        self, kwargs: Dict[str, Any], flags: RemoteMethodFlags
    ):
        # in single-threaded mode, we're going to remain synchronous, so call
        # `run`, not `start`, and return an actual result.
        # note this shouldn't call self.run() as that has different semantics
        # (we want errors to raise)
        if self.process is None:  # mypy appeasement
            raise dbt.exceptions.InternalException(
                'Cannot run a None process'
            )
        self.process.task_exec()
        with StateHandler(self):
            self.result = self.get_result()
        return self.result

def generate_base(model, model_dict, config, manifest, source_config,
                  provider, adapter=None):
    """Generate the common aspects of the config dict."""
    if provider is None:
        raise dbt.exceptions.InternalException(
            "Invalid provider given to context: {}".format(provider))

    target_name = config.target_name
    target = config.to_profile_info()
    del target['credentials']
    target.update(config.credentials.serialize(with_aliases=True))
    target['type'] = config.credentials.type
    target.pop('pass', None)
    target['name'] = target_name

    adapter = get_adapter(config)

    context = {'env': target}

    pre_hooks = None
    post_hooks = None

def handle_completed(self):
        # killed handlers don't get a result.
        if self.handler.state != TaskHandlerState.Killed:
            if self.handler.result is None:
                # there wasn't an error before, but there sure is one now
                self.handler.error = dbt_error(
                    dbt.exceptions.InternalException(
                        'got an invalid result=None, but state was {}'
                        .format(self.handler.state)
                    )
                )
            elif self.handler.task.interpret_results(self.handler.result):
                self.handler.state = TaskHandlerState.Success
            else:
                self.handler.state = TaskHandlerState.Failed
        self.set_end()

def recursively_prepend_ctes(model, manifest):
    if model.extra_ctes_injected:
        return (model, model.extra_ctes, manifest)

    if dbt.flags.STRICT_MODE:
        if not isinstance(model, tuple(COMPILED_TYPES.values())):
            raise dbt.exceptions.InternalException(
                'Bad model type: {}'.format(type(model))
            )

    prepended_ctes = []

    for cte in model.extra_ctes:
        cte_id = cte.id
        cte_to_add = manifest.nodes.get(cte_id)
        cte_to_add, new_prepended_ctes, manifest = recursively_prepend_ctes(
            cte_to_add, manifest)
        _extend_prepended_ctes(prepended_ctes, new_prepended_ctes)
        new_cte_name = '__dbt__CTE__{}'.format(cte_to_add.name)
        sql = ' {} as (\n{}\n)'.format(new_cte_name, cte_to_add.compiled_sql)
        _add_prepended_cte(prepended_ctes, InjectedCTE(id=cte_id, sql=sql))

    model.prepend_ctes(prepended_ctes)

def __post_init__(self):
        if self.seed_file_path == '':
            raise dbt.exceptions.InternalException(
                'Seeds should always have a seed_file_path'
            )

"""Wait for results off the queue. If there is a timeout set, and it is
        exceeded, raise an RPCTimeoutException.
        """
        while True:
            get_timeout = self._next_timeout()
            try:
                msgtype, value = self.queue.get(timeout=get_timeout)
            except QueueEmpty:
                raise dbt.exceptions.RPCTimeoutException(self.timeout)

            if msgtype == QueueMessageType.Log:
                self.logs.append(value)
            elif msgtype in QueueMessageType.terminating():
                return msgtype, value
            else:
                raise dbt.exceptions.InternalException(
                    'Got invalid queue message type {}'.format(msgtype)
                )

def method(self) -> str:
        if self.task.METHOD_NAME is None:  # mypy appeasement
            raise dbt.exceptions.InternalException(
                f'In the request handler, got a task({self.task}) with no '
                'METHOD_NAME'
            )
        return self.task.METHOD_NAME

def handle_exception(self, e, ctx):
        catchable_errors = (CompilationException, RuntimeException)
        if isinstance(e, catchable_errors):
            error = self._handle_catchable_exception(e, ctx)
        elif isinstance(e, InternalException):
            error = self._handle_internal_exception(e, ctx)
        else:
            error = self._handle_generic_exception(e, ctx)
        return error