Method _signal_stop_worker

psycopg_pool/psycopg_pool/pool.py:468–481

(self)

            pos.fail(PoolClosed(f"the pool {self.name!r} is closed"))

    def _signal_stop_worker(self) -> list[Worker]:
        # Stop the scheduler
        self._sched.enter(0, None)

        # Stop the worker tasks
        workers, self._workers = (self._workers[:], [])
        for _ in workers:
            self.run_task(StopWorker(self))

        if self._sched_runner:  # likely
            workers.append(self._sched_runner)
            self._sched_runner = None

        return workers

    def __enter__(self) -> Self:
        self._open_implicit = False