diff --git a/queue_job/jobrunner/channels.py b/queue_job/jobrunner/channels.py index c895d9caf3..5a82145c8e 100644 --- a/queue_job/jobrunner/channels.py +++ b/queue_job/jobrunner/channels.py @@ -404,7 +404,16 @@ class Channel: without risking to overflow the system. """ - def __init__(self, name, parent, capacity=None, sequential=False, throttle=0): + def __init__( + self, + name, + parent, + capacity=None, + sequential=False, + throttle=0, + capacity_default=None, + sequential_default=False, + ): self.name = name self.parent = parent if self.parent: @@ -414,9 +423,13 @@ def __init__(self, name, parent, capacity=None, sequential=False, throttle=0): self._running = set() self._failed = set() self._pause_until = 0 # utc seconds since the epoch - self.capacity = capacity + self.sequential = sequential or (parent and parent.sequential_default) + self.sequential_default = sequential_default + self.capacity = ( + capacity if (capacity is not None) else (parent and parent.capacity_default) + ) + self.capacity_default = capacity_default self.throttle = throttle # seconds - self.sequential = sequential @property def sequential(self): @@ -432,12 +445,16 @@ def configure(self, config): Supported keys are: * capacity + * capacity_default (default for sub channels) * sequential + * sequential_default (default for sub channels) * throttle """ assert self.fullname.endswith(config["name"]) self.capacity = config.get("capacity", None) + self.capacity_default = config.get("capacity_default", None) self.sequential = bool(config.get("sequential", False)) + self.sequential_default = config.get("sequential_default", False) self.throttle = int(config.get("throttle", 0)) if self.sequential and self.capacity != 1: raise ValueError("A sequential channel must have a capacity of 1") @@ -866,22 +883,23 @@ def parse_simple_config(cls, config_string): continue config = {} config_items = split_strip(channel_config_string, ":") - name = config_items[0] - if not name: + if not (name := config_items.pop(0)): raise ValueError( f"Invalid channel config {config_string}: missing channel name" ) config["name"] = name - if len(config_items) > 1: - capacity = config_items[1] + if len(config_items) > 0: try: - config["capacity"] = int(capacity) + config["capacity"] = int(config_items[0]) + config_items.pop(0) except Exception as ex: - raise ValueError( - f"Invalid channel config {config_string}: " - f"invalid capacity {capacity}" - ) from ex - for config_item in config_items[2:]: + if name == "root": + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity {config_items[0]}" + ) from ex + + for config_item in config_items: kv = split_strip(config_item, "=") if len(kv) == 1: k, v = kv[0], True @@ -897,7 +915,16 @@ def parse_simple_config(cls, config_string): f"Invalid channel config {config_string}: " f"duplicate key {k}" ) - config[k] = v + if k == "capacity_default": + try: + config[k] = int(v) + except Exception as ex: + raise ValueError( + f"Invalid channel config {config_string}: " + f"invalid capacity_default {v}" + ) from ex + else: + config[k] = v else: config["capacity"] = 1 res.append(config) @@ -910,6 +937,17 @@ def simple_configure(self, config_string): >>> c = cm.get_channel_by_name('root') >>> c.capacity 1 + + >>> cm.simple_configure('root:bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:bogus: invalid capacity bogus + + >>> cm.simple_configure('root:4,:2') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4,:2: missing channel name + >>> cm.simple_configure('root:4,autosub.sub:2,seq:1:sequential') >>> cm.get_channel_by_name('root').capacity 4 @@ -926,7 +964,28 @@ def simple_configure(self, config_string): 1 >>> cm.get_channel_by_name('seq').sequential True - """ + + >>> cm.simple_configure('root:4:capacity_default=bogus') + Traceback (most recent call last): + ... + ValueError: Invalid channel config root:4:capacity_default=bogus: invalid capacity_default bogus + + >>> cm.simple_configure('root:4,sub:3:capacity_default=2') + >>> cm.get_channel_by_name('root.sub').capacity + 3 + >>> cm.get_channel_by_name('root.sub.auto', autocreate=True).capacity + 2 + + >>> cm.simple_configure('root:4,seq:2:sequential') + Traceback (most recent call last): + ... + ValueError: A sequential channel must have a capacity of 1 + + >>> cm.simple_configure('root:4,seq:sequential_default') + >>> cm.get_channel_by_name('root.seq.auto', autocreate=True).sequential + True + + """ # noqa: E501 for config in ChannelManager.parse_simple_config(config_string): self.get_channel_from_config(config) diff --git a/queue_job/tests/test_runner_channels.py b/queue_job/tests/test_runner_channels.py index b00dd26261..313e350bd3 100644 --- a/queue_job/tests/test_runner_channels.py +++ b/queue_job/tests/test_runner_channels.py @@ -12,6 +12,7 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( channels, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") diff --git a/queue_job/tests/test_runner_runner.py b/queue_job/tests/test_runner_runner.py index 1adfa790dd..04c3fc1946 100644 --- a/queue_job/tests/test_runner_runner.py +++ b/queue_job/tests/test_runner_runner.py @@ -13,9 +13,10 @@ @tagged("doctest") class TestDoctest(BaseCase): def test_doctest(self): - doctest.testmod( + results = doctest.testmod( runner, exclude_empty=True, optionflags=doctest.REPORT_ONLY_FIRST_FAILURE ) + self.assertEqual(results.failed, 0, "doctest failed") @tagged("-at_install", "post_install")