--postpone-jobs flag (#146)

Author: droserasprout

CHANGELOG.md

@@ -5,7 +5,9 @@
 ### Added
 
 * Human-readable `CHANGELOG.md` 🕺
-* `--forbid-reindexing` option added to `dipdup run` command. If this flag is set, DipDup will raise `ReindexingRequiredError` when reindexing is triggered for any reason. A database won't be truncated. To continue indexing with existing database run `UPDATE dipdup_schema SET reindex = NULL;`
+* Two new options added to `dipdup run` command:
+  * `--forbid-reindexing` – raise `ReindexingRequiredError` instead of truncating database when reindexing is triggered for any reason. To continue indexing with existing database run `UPDATE dipdup_schema SET reindex = NULL;`
+  * `--postpone-jobs` – job scheduler won't start until all indexes are synchronized. 
 
 ### Changed
 

src/dipdup/cli.py

@@ -136,17 +136,24 @@ async def cli(ctx, config: List[str], env_file: List[str], logging_config: str):
 @cli.command(help='Run indexing')
 @click.option('--reindex', is_flag=True, help='Drop database and start indexing from scratch')
 @click.option('--oneshot', is_flag=True, help='Synchronize indexes wia REST and exit without starting WS connection')
+@click.option('--postpone-jobs', is_flag=True, help='Do not start job scheduler until all indexes are synchronized')
 @click.option('--forbid-reindexing', is_flag=True, help='Raise exception instead of truncating database when reindexing is triggered')
 @click.pass_context
 @cli_wrapper
-async def run(ctx, reindex: bool, oneshot: bool, forbid_reindexing: bool) -> None:
+async def run(
+    ctx,
+    reindex: bool,
+    oneshot: bool,
+    postpone_jobs: bool,
+    forbid_reindexing: bool,
+) -> None:
     config: DipDupConfig = ctx.obj.config
     config.initialize()
     set_decimal_context(config.package)
     if forbid_reindexing:
         context.forbid_reindexing = True
     dipdup = DipDup(config)
-    await dipdup.run(reindex, oneshot)
+    await dipdup.run(reindex, oneshot, postpone_jobs)
 
 
 @cli.command(help='Generate missing callbacks and types')

src/dipdup/dipdup.py

@@ -225,7 +225,7 @@ async def init(self, overwrite_types: bool = True) -> None:
     async def docker_init(self, image: str, tag: str, env_file: str) -> None:
         await self._codegen.docker_init(image, tag, env_file)
 
-    async def run(self, reindex: bool, oneshot: bool) -> None:
+    async def run(self, reindex: bool, oneshot: bool, postpone_jobs: bool) -> None:
         """Run indexing process"""
         tasks: Set[Task] = set()
         async with AsyncExitStack() as stack:
@@ -242,6 +242,8 @@ async def run(self, reindex: bool, oneshot: bool) -> None:
             start_scheduler_event: Optional[Event] = None
             if not oneshot:
                 start_scheduler_event = await self._set_up_scheduler(stack, tasks)
+                if not postpone_jobs:
+                    start_scheduler_event.set()
                 spawn_datasources_event = await self._spawn_datasources(tasks)
 
             for name in self._config.indexes: