Skip to content

Commit 48b9537

Browse files
committed
fix(util): replay worker warnings from a quiet parallel batch
- parallelize discarded every buffered record when flush_at_end was off - records at WARNING and above are now replayed either way - printed output and records below WARNING still obey flush_at_end - also recovers the _logger.error calls in OSW.delete_entity's workers
1 parent b097b10 commit 48b9537

2 files changed

Lines changed: 75 additions & 12 deletions

File tree

‎src/osw/utils/util.py‎

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -412,10 +412,12 @@ def parallelize(
412412
iterable:
413413
The iterable, who's items are to be passed as singles to the function.
414414
flush_at_end:
415-
If True, what the tasks print is written to stdout in the order of
416-
``iterable`` once the batch has finished. If False, it is discarded.
417-
Either way it is captured while the tasks run, so that concurrent
418-
printing cannot garble the progress bar.
415+
If True, what the tasks print and log is written out in the order of
416+
``iterable`` once the batch has finished. If False, it is discarded,
417+
except for log records at WARNING and above. Those report a problem
418+
rather than progress, so they are written out either way. Either way
419+
the output is captured while the tasks run, so that concurrent writing
420+
cannot garble the progress bar.
419421
progress_bar:
420422
If True, a progress bar will be displayed.
421423
return_exceptions:
@@ -502,13 +504,16 @@ async def _capture(task):
502504
sys.stdout = stdout.stream
503505
osw_logger.handlers = saved_handlers
504506
osw_logger.propagate = saved_propagate
505-
if flush_at_end:
506-
# in the finally block, so a failed batch still reports what its
507-
# tasks had to say
508-
for output, task_records in zip(outputs, records):
509-
if output:
510-
print(output, end="")
511-
for record in task_records:
507+
# in the finally block, so a failed batch still reports what its
508+
# tasks had to say
509+
for output, task_records in zip(outputs, records):
510+
if flush_at_end and output:
511+
print(output, end="")
512+
for record in task_records:
513+
# A warning reports a problem, not progress, so a batch asked
514+
# to stay quiet still passes it on. Discarding it would hide
515+
# from the caller that something went wrong in a worker.
516+
if flush_at_end or record.levelno >= logging.WARNING:
512517
log_handler.replay(record)
513518

514519
return results

‎tests/test_logging_setup.py‎

Lines changed: 59 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -289,6 +289,12 @@ def _log_item(item):
289289
return item
290290

291291

292+
def _warn_on_item(item):
293+
logging.getLogger(WORKER).info(f"handled {item}")
294+
logging.getLogger(WORKER).warning(f"trouble with {item}")
295+
return item
296+
297+
292298
def test_worker_records_are_replayed_in_input_order(collected):
293299
"""The tasks finish in whatever order they finish, so the replay is what
294300
puts the messages back in the order of the iterable."""
@@ -303,12 +309,64 @@ def test_worker_records_are_replayed_in_input_order(collected):
303309

304310

305311
def test_worker_records_are_dropped_without_flush_at_end(collected):
306-
"""flush_at_end=False means quiet, which is what debug=False asks for."""
312+
"""flush_at_end=False means quiet for progress, which is what debug=False
313+
asks for."""
307314
parallelize(_log_item, [1, 2, 3], flush_at_end=False, progress_bar=False)
308315

309316
assert collected.messages(WORKER) == []
310317

311318

319+
def test_worker_warnings_survive_a_quiet_batch(collected):
320+
"""A warning reports a problem rather than progress, so a batch asked to
321+
stay quiet still passes it on. These call sites used warnings.warn before,
322+
which parallelize never suppressed, so dropping them would lose output a
323+
caller has today."""
324+
parallelize(_warn_on_item, [1, 2], flush_at_end=False, progress_bar=False)
325+
326+
assert collected.messages(WORKER) == ["trouble with 1", "trouble with 2"]
327+
328+
329+
def test_worker_errors_survive_a_quiet_batch(collected):
330+
"""Errors are replayed on the same grounds as warnings. OSW.delete_entity
331+
logs one from inside a worker it parallelizes, at src/osw/core.py."""
332+
333+
def _fail_quietly(item):
334+
logging.getLogger(WORKER).error(f"could not handle {item}")
335+
336+
parallelize(_fail_quietly, [1], flush_at_end=False, progress_bar=False)
337+
338+
assert collected.messages(WORKER) == ["could not handle 1"]
339+
340+
341+
def test_worker_warnings_reach_pytests_caplog(caplog, osw_logger):
342+
"""caplog puts its handler on the root logger, so a replayed record has to
343+
travel the same path as any other to be seen. The integration test
344+
tests/integration/performance_test.py asserts on a warning from a parallel
345+
get_page batch this way, and it needs wiki credentials to run."""
346+
caplog.set_level(logging.WARNING, logger="osw")
347+
348+
parallelize(_warn_on_item, [1], flush_at_end=False, progress_bar=False)
349+
350+
assert any("trouble with 1" in record.message for record in caplog.records)
351+
352+
353+
def test_worker_warnings_are_replayed_in_input_order(collected):
354+
"""The tasks finish in whatever order they finish, so the warnings are put
355+
back in the order of the iterable like everything else."""
356+
parallelize(_warn_on_item, [1, 2, 3, 4], flush_at_end=True, progress_bar=False)
357+
358+
assert collected.messages(WORKER) == [
359+
"handled 1",
360+
"trouble with 1",
361+
"handled 2",
362+
"trouble with 2",
363+
"handled 3",
364+
"trouble with 3",
365+
"handled 4",
366+
"trouble with 4",
367+
]
368+
369+
312370
def test_worker_records_reach_the_application_handler(osw_logger):
313371
"""The capture replays into whatever the handler chain holds, so a batch's
314372
records land in the application's handlers like any other."""

0 commit comments

Comments
 (0)