I found a workaround to solve this problem. This issue only affects threads created from the main thread. If you create a thread inside the server thread pool (e.g. at the first network request), it works as expected. So, lazy thread initialization is enough.
Code with workaround:
[details=Show More]
import functools import logging import queue import threading from typing import Callable, Optional from flask import Flask, Request, Response, request logger: logging.Logger = logging.getLogger('thread_test') logger.setLevel(logging.DEBUG) handler = logging.StreamHandler() formatter = logging.Formatter("%(asctime)s [%(levelname)s] " "%(filename)s:%(lineno)s %(funcName)s(): " "'%(message)s'") handler.setFormatter(formatter) logger.addHandler(handler) class Scheduler: class ScheduleThread(threading.Thread): def __init__(self, interval: float = 10.0) -> None: super().__init__() self.name += " (Scheduler)" self._interval = interval self.job_queue = queue.Queue() self.event_job = threading.Event() self.event_stop = threading.Event() def run(self) -> None: while not self.event_stop.is_set(): if self.event_job.wait(self._interval): self.event_job.clear() logger.debug(id(self.job_queue)) logger.debug(self.job_queue.__dict__) while not self.job_queue.empty(): logger.debug("") job = self.job_queue.get(False) logger.debug(job) if isinstance(job, functools.partial): logger.debug("") job() logger.debug("") def start(self) -> None: logger.debug("") self._thread = self.ScheduleThread() self._thread.start() def add_job(self, job_func: Callable, *args, **kwargs) -> None: if self._thread is None: self.start() if self._thread is None: return logger.debug("") job = functools.partial(job_func, *args, **kwargs) logger.debug(id(self._thread.job_queue)) logger.debug(self._thread.job_queue.__dict__) self._thread.job_queue.put(job) logger.debug(self._thread.job_queue.__dict__) self._thread.event_job.set() logger.debug("") def __init__(self) -> None: self._thread: Optional[Scheduler.ScheduleThread] = None def __del__(self) -> None: if self._thread is not None: self._thread.event_stop.set() flask_app = Flask("Thread Test") scheduler: Optional[Scheduler] = None with flask_app.app_context(): scheduler = Scheduler() def test_func() -> None: logger.debug("Thread: %s", threading.current_thread().name) # Cloud Functions def app_test(_req: Request) -> Response: """HTTP Cloud Function. Args: req (flask.Request): The request object.