text: # tasks.py from https://github.com/dmontagu/fastapi-utils
  import asyncio
  import logging
  from asyncio import ensure_future
  from functools import wraps
  from traceback import format_exception
  from typing import Any, Callable, Coroutine, Optional, Union
  
  from starlette.concurrency import run_in_threadpool
  
  from constants import GJS_PREFIX
  
  NoArgsNoReturnFuncT = Callable[[], None]
  NoArgsNoReturnAsyncFuncT = Callable[[], Coroutine[Any, Any, None]]
  NoArgsNoReturnDecorator = Callable[
      [Union[NoArgsNoReturnFuncT, NoArgsNoReturnAsyncFuncT]], NoArgsNoReturnAsyncFuncT
  ]
  
  
  def repeat_every(
      *,
      seconds: float,
      wait_first: bool = False,
      logger: Optional[logging.Logger] = None,
      raise_exceptions: bool = False,
      max_repetitions: Optional[int] = None
  ) -> NoArgsNoReturnDecorator:
      """
      Returns decorator that modifies a function to be run periodically after it's first call
  
      Parameters
      ----------
      seconds: float
          Repeat every x seconds
      wait_first: bool (default False)
          Wait before executing first call
      logger: Optional[logging.Logger] (default None)
          log any exceptions raised
      raise_exceptions: bool (default False)
          Only use if you want periodic execution to stop if there's an exception.
          Errors raised by the deecorated function will be raised to the  event loops exception handler.
      max_repetitions: Optional[int] (default None)
          Max number of times to repeat function, otherwise run forever
      """
  
      def decorator(
          func: Union[NoArgsNoReturnAsyncFuncT, NoArgsNoReturnFuncT]
      ) -> NoArgsNoReturnAsyncFuncT:
          """
          Convert decorated function to repeated function
          """
          is_coroutine = asyncio.iscoroutinefunction(func)
  
          @wraps(func)
          async def wrapped() -> None:
              repetitions = 0
  
              async def loop() -> None:
                  nonlocal repetitions
                  if wait_first:
                      await asyncio.sleep(seconds)
                  while max_repetitions is None or repetitions < max_repetitions:
                      try:
                          if is_coroutine:
                              await func()
                          else:
                              await run_in_threadpool(func)
                          repetitions += 1
                      except Exception as exc:
                          if logger is not None:
                              formatted_exception = "".join(
                                  format_exception(type(exc), exc, exc.__traceback__)
                              )
                              logger.error(formatted_exception)
                          if raise_exceptions:
                              raise exc
                      await asyncio.sleep(seconds)
  
              ensure_future(loop())
  
          return wrapped
  
      return decorator
  
  
  def prefix(dictionary):
      """
      Add prefix to assets, html, css, components, styles keys
  
      Parameters
      ----------
      dictionary: Dict
          Dictionary object to filter
      """
      
      return {
          **dictionary,
          **{
              f"{GJS_PREFIX}{k}": v
              for k, v in dictionary.items()
              if k == "assets"
              or k == "html"
              or k == "css"
              or k == "components"
              or k == "pages"
              or k == "styles"
          },
      }