INNER CODE UNIT · Python

cleanup

stacklok/codegate · src/codegate/cli.py:74

    async def cleanup(self) -> None:
        """Cleanup server resources and ensure graceful shutdown."""
        self.logger.debug("Cleaning up server resources")
        if not self._should_exit:
            self._should_exit = True
            self.logger.debug("Initiating server shutdown")
            self._shutdown_event.set()

            if hasattr(self.server, "shutdown"):
                self.logger.debug("Shutting down server")
                await self.server.shutdown()

            # Ensure all connections are closed
            tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
            [task.cancel() for task in tasks]

            await asyncio.gather(*tasks, return_exceptions=True)
            self.logger.debug("Server shutdown complete")

View source record →