Enqueue messages to redis with backfill command

cli.py

@@ -3,6 +3,7 @@ import os
 import sys
 
 import click
+from worker import inspect_many_blocks_task
 
 from mev_inspect.concurrency import coro
 from mev_inspect.crud.prices import write_prices
@@ -91,6 +92,17 @@ async def inspect_many_blocks_command(
     )
 
 
+@cli.command()
+@click.argument("after_block", type=int)
+@click.argument("before_block", type=int)
+@click.argument("batch_size", type=int, default=10)
+def enqueue_many_blocks_command(after_block: int, before_block: int, batch_size: int):
+    for batch_after_block in range(after_block, before_block, batch_size):
+        batch_before_block = min(batch_after_block + batch_size, before_block)
+        logger.info(f"Sending {batch_after_block} to {batch_before_block}")
+        inspect_many_blocks_task.send(batch_after_block, batch_before_block)
+
+
 @cli.command()
 @coro
 async def fetch_all_prices():

mev

@@ -45,10 +45,9 @@ case "$1" in
   backfill)
         start_block_number=$2
         end_block_number=$3
-        n_workers=$4
 
-        echo "Backfilling from $start_block_number to $end_block_number with $n_workers workers"
-        poetry run python backfill.py $start_block_number $end_block_number $n_workers
+        echo "Backfilling from $start_block_number to $end_block_number"
+        kubectl exec -ti deploy/mev-inspect -- poetry run enqueue-many-blocks $start_block_number $end_block_number
 	;;
   inspect)
         block_number=$2

pyproject.toml

@@ -35,6 +35,7 @@ build-backend = "poetry.core.masonry.api"
 [tool.poetry.scripts]
 inspect-block = 'cli:inspect_block_command'
 inspect-many-blocks = 'cli:inspect_many_blocks_command'
+enqueue-many-blocks = 'cli:enqueue_many_blocks_command'
 fetch-block = 'cli:fetch_block_command'
 fetch-all-prices = 'cli:fetch_all_prices'
 

worker.py

@@ -36,8 +36,8 @@ dramatiq.set_broker(broker)
 @contextmanager
 def session_scope(Session=None):
     if Session is None:
-        return None
-
+        yield None
+    else:
         with Session() as session:
             yield session