ClickUp Operator

from __future__ import annotations import asyncio import logging import signal import sys import warnings from typing import Any from gunicorn.arbiter import Arbiter from gunicorn.workers.base import Worker from uvicorn.config import Config from uvicorn.main import Server warnings.warn( "The `uvicorn.workers` module is deprecated. Please use `uvicorn-worker` package instead.\n" "For more details, see https://github.com/Kludex/uvicorn-worker.", DeprecationWarning, ) class UvicornWorker(Worker): """ A worker class for Gunicorn that interfaces with an ASGI consumer callable, rather than a WSGI callable. """ CONFIG_KWARGS: dict[str, Any] = {"loop": "auto", "http": "auto"} def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) logger = logging.getLogger("uvicorn.error") logger.handlers = self.log.error_log.handlers logger.setLevel(self.log.error_log.level) logger.propagate = False logger = logging.getLogger("uvicorn.access") logger.handlers = self.log.access_log.handlers logger.setLevel(self.log.access_log.level) logger.propagate = False config_kwargs: dict = { "app": None, "log_config": None, "timeout_keep_alive": self.cfg.keepalive, "timeout_notify": self.timeout, "callback_notify": self.callback_notify, "limit_max_requests": self.max_requests, "forwarded_allow_ips": self.cfg.forwarded_allow_ips, } if self.cfg.is_ssl: ssl_kwargs = { "ssl_keyfile": self.cfg.ssl_options.get("keyfile"), "ssl_certfile": self.cfg.ssl_options.get("certfile"), "ssl_keyfile_password": self.cfg.ssl_options.get("password"), "ssl_version": self.cfg.ssl_options.get("ssl_version"), "ssl_cert_reqs": self.cfg.ssl_options.get("cert_reqs"), "ssl_ca_certs": self.cfg.ssl_options.get("ca_certs"), "ssl_ciphers": self.cfg.ssl_options.get("ciphers"), } config_kwargs.update(ssl_kwargs) if self.cfg.settings["backlog"].value: config_kwargs["backlog"] = self.cfg.settings["backlog"].value config_kwargs.update(self.CONFIG_KWARGS) self.config = Config(**config_kwargs) def init_process(self) -> None: self.config.setup_event_loop() super().init_process() def init_signals(self) -> None: # Reset signals so Gunicorn doesn't swallow subprocess return codes # other signals are set up by Server.install_signal_handlers() # See: https://github.com/encode/uvicorn/issues/894 for s in self.SIGNALS: signal.signal(s, signal.SIG_DFL) signal.signal(signal.SIGUSR1, self.handle_usr1) # Don't let SIGUSR1 disturb active requests by interrupting system calls signal.siginterrupt(signal.SIGUSR1, False) def _install_sigquit_handler(self) -> None: """Install a SIGQUIT handler on workers. - https://github.com/encode/uvicorn/issues/1116 - https://github.com/benoitc/gunicorn/issues/2604 """ loop = asyncio.get_running_loop() loop.add_signal_handler(signal.SIGQUIT, self.handle_exit, signal.SIGQUIT, None) async def _serve(self) -> None: self.config.app = self.wsgi server = Server(config=self.config) self._install_sigquit_handler() await server.serve(sockets=self.sockets) if not server.started: sys.exit(Arbiter.WORKER_BOOT_ERROR) def run(self) -> None: return asyncio.run(self._serve()) async def callback_notify(self) -> None: self.notify() class UvicornH11Worker(UvicornWorker): CONFIG_KWARGS = {"loop": "asyncio", "http": "h11"}