How to use the dramatiq.brokers.stub.StubBroker function in dramatiq

import apistar_dramatiq
import dramatiq

from apistar import App, Component, Route
from apistar_dramatiq import actor
from dramatiq.brokers.stub import StubBroker

broker = StubBroker()
broker.emit_after("process_boot")
dramatiq.set_broker(broker)


class Settings(dict):
    pass


class MissingDep:
    pass


class SettingsComponent(Component):
    def resolve(self) -> Settings:
        return {"EXAMPLE": 42}

def test_shutdown_notifications_platform_not_supported(recwarn, monkeypatch):
    # monkeypatch fake platform to test logging.
    monkeypatch.setattr(shutdown, "current_platform", "not supported")

    # Given a broker configured with the shutdown notifier
    broker = StubBroker(middleware=[shutdown.ShutdownNotifications()])

    # When the process boots
    broker.emit_after("process_boot")

    # A platform support warning is issued
    assert len(recwarn) == 1
    assert str(recwarn[0].message) == ("ShutdownNotifications cannot kill threads "
                                       "on your current platform ('not supported').")

def app_with_scout(config=None):
    """
    Context manager that configures a Dramatiq app with Scout middleware
    installed.
    """
    # Enable Scout by default in tests.
    if config is None:
        config = {"monitor": True}

    # Disable running the agent.
    config["core_agent_launch"] = False

    broker = StubBroker()
    broker.emit_after("process_boot")
    dramatiq.set_broker(broker)

    @dramatiq.actor(max_retries=0)
    def hello():
        return "Hello World!"

    @dramatiq.actor(max_retries=0)
    def fail():
        raise ValueError("BØØM!")  # non-ASCII

    worker = dramatiq.Worker(broker, worker_timeout=0)

    # Setup according to https://docs.scoutapm.com/#dramatiq
    Config.set(**config)
    broker.add_middleware(ScoutMiddleware(), before=broker.middleware[0].__class__)

def test_shutdown_notifications_worker_shutdown_messages(raise_thread_exception, caplog):
    # capture all messages
    caplog.set_level(logging.NOTSET)

    # Given a middleware with two "threads"
    middleware = shutdown.ShutdownNotifications()
    middleware.notifications = [1, 2]

    # Given a broker configured with the shutdown notifier
    broker = StubBroker(middleware=[middleware])

    # When the worker is shutdown
    broker.emit_before("worker_shutdown", None)

    # Shutdown interrupts are raised in both threads
    raise_thread_exception.assert_has_calls([
        mock.call(1, shutdown.Shutdown),
        mock.call(2, shutdown.Shutdown),
    ])

    # And shutdown notifications are logged
    assert len(caplog.record_tuples) == 3
    assert caplog.record_tuples == [
        ("dramatiq.middleware.shutdown.ShutdownNotifications", logging.DEBUG, (
            "Sending shutdown notification to worker threads..."
        )),

host = host or None
                port = int(port) if port else None
        else:
            kind = self.DRAMATIQ_BROKER_TYPE
            host = self.DRAMATIQ_BROKER_HOST or None
            port = self.DRAMATIQ_BROKER_PORT or None

        # Separate non-null args
        kwargs = [("host", host), ("port", port)]
        kwargs = {k: v for k, v in kwargs if v is not None}

        # Initializes broker
        if kind == "stub":
            from dramatiq.brokers.stub import StubBroker

            broker = StubBroker()
        elif kind == "redis":
            from dramatiq.brokers.redis import RedisBroker

            broker = RedisBroker(**kwargs)
        elif kind == "rabbitmq":
            from dramatiq.brokers.rabbitmq import RabbitmqBroker

            broker = RabbitmqBroker(**kwargs)
        else:
            raise ValueError(f"invalid dramatiq broker: {kind}")

        # Configure as default and exit
        dramatiq.set_broker(broker)
        return broker

import dramatiq
from dramatiq.brokers.stub import StubBroker


from solarforecastarbiter.validation import tasks as validation_tasks


if 'REDIS_URL' in os.environ:  # pragma: no cover
    from dramatiq.brokers.redis import RedisBroker
    broker = RedisBroker(url=os.environ['REDIS_URL'],
                         db=0,
                         namespace='sfa-queue')
else:
    broker = StubBroker()
dramatiq.set_broker(broker)


def enqueue_function(func, *args, **kwargs):
    """Convience function to queue function. Will allow for altering task
    queue without changing code that queues up the tasks. If broker
    is not a StubBroker, the task is sent to the broker. Otherwise
    the task is commputed synchronously"""
    if isinstance(broker, StubBroker):
        return func(*args, **kwargs)
    else:
        return func.send(*args, **kwargs)


@dramatiq.actor(max_retries=3)
def immediate_observation_validation(*args, **kwargs):

def enqueue_function(func, *args, **kwargs):
    """Convience function to queue function. Will allow for altering task
    queue without changing code that queues up the tasks. If broker
    is not a StubBroker, the task is sent to the broker. Otherwise
    the task is commputed synchronously"""
    if isinstance(broker, StubBroker):
        return func(*args, **kwargs)
    else:
        return func.send(*args, **kwargs)

import os

from dramatiq.brokers.redis import RedisBroker
from dramatiq.brokers.stub import StubBroker
from flask import current_app

if os.getenv('APP_SETTINGS') == 'fittrackee_api.config.TestingConfig':
    broker = StubBroker
else:
    broker = RedisBroker


class BaseConfig:
    """Base configuration"""

    DEBUG = False
    TESTING = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    BCRYPT_LOG_ROUNDS = 13
    TOKEN_EXPIRATION_DAYS = 30
    TOKEN_EXPIRATION_SECONDS = 0
    PASSWORD_TOKEN_EXPIRATION_SECONDS = 3600
    UPLOAD_FOLDER = os.path.join(current_app.root_path, 'uploads')
    PICTURE_ALLOWED_EXTENSIONS = {'jpg', 'png', 'gif'}

from . import db

            await db.db.set_bind(db.CONNECTION_STR)

        loop.run_until_complete(run())

    def before_worker_shutdown(self, broker, worker):
        shutdown.set()


MIDDLEWARE = [
    m() for m in (AgeLimit, ShutdownNotifications, Callbacks, Pipelines, Retries)
]

if TEST:
    broker = StubBroker(middleware=MIDDLEWARE)
    broker.emit_after("process_boot")
else:
    broker = RedisBroker(
        connection_pool=redis.ConnectionPool.from_url(REDIS_HOST), middleware=MIDDLEWARE
    )

broker.add_middleware(InitDB())
dramatiq.set_broker(broker)