def drop()

in databao/core/thread.py [0:0]

• 24 lines of code
• 6 McCabe index (conditional complexity)

    def drop(self, n: int = 1) -> None:
        """Remove N last user queries from this thread along with the answer it produced."""
        sum_, n_groups = 0, 0
        for group in reversed(self._opas):
            sum_ += len(group)
            n_groups += 1
            if sum_ >= n:
                break

        n_materialized_group = n_groups - (len(self._opas) - self._opas_processed_count)

        # We need to drop `n` individual opas, combined into `n_groups` groups,
        # `n_materialized_group` of which are materialized.
        if sum_ == n:
            # Full drop of groups
            self._opas = self._opas[:-n_groups]
        else:
            full_groups = n_groups - 1
            if full_groups > 0:
                self._opas = self._opas[:-full_groups]
            self._opas[-1] = self._opas[-1][: -(sum_ - n)]

        self._agent.executor.drop_last_opa_group(self._agent.cache.scoped(self._cache_scope), n=n_materialized_group)
        self._opas_processed_count -= n_materialized_group

        if self._opas:
            print(
                f"Dropped last {n} operation{'s' if n > 1 else ''}. Last remaining operation:"
                f"\n{self._opas[-1][-1].query}"
            )
        else:
            print("Dropped all operations.")