Add assert concurrency configuration

spectacles/cli.py

@@ -27,6 +27,7 @@
 from spectacles.logger import set_file_handler
 from spectacles.runner import Runner
 from spectacles.utils import log_duration
+from spectacles.validators.data_test import QUERY_SLOT_LIMIT
 
 __version__ = importlib.metadata.version("spectacles")
 
@@ -341,6 +342,7 @@ def main() -> None:
                 api_version=args.api_version,
                 remote_reset=args.remote_reset,
                 pin_imports=pin_imports,
+                concurrency=args.concurrency,
             )
         )
     elif args.command == "content":
@@ -709,6 +711,16 @@ def _build_assert_subparser(
     _build_validator_subparser(subparser_action, subparser)
     _build_select_subparser(subparser_action, subparser)
 
+    subparser.add_argument(
+        "--concurrency",
+        type=int,
+        default=QUERY_SLOT_LIMIT,
+        help=(
+            "Specify the number of concurrent queries you want to have running "
+            f"against your data warehouse. The default is {QUERY_SLOT_LIMIT}."
+        ),
+    )
+
 
 def _build_content_subparser(
     subparser_action: argparse._SubParsersAction,  # type: ignore[type-arg]
@@ -896,6 +908,7 @@ async def run_assert(
     api_version: float,
     remote_reset: bool,
     pin_imports: Dict[str, str],
+    concurrency: int,
 ) -> None:
     # Don't trust env to ignore .netrc credentials
     async_client = httpx.AsyncClient(trust_env=False)
@@ -905,7 +918,9 @@ async def run_assert(
         )
         runner = Runner(client, project, remote_reset, pin_imports)
 
-        results = await runner.validate_data_tests(ref, filters)
+        results = await runner.validate_data_tests(
+            ref, filters, query_slot_limit=concurrency
+        )
     finally:
         await async_client.aclose()
 

spectacles/runner.py

@@ -17,6 +17,7 @@
     LookMLValidator,
     SqlValidator,
 )
+from spectacles.validators.data_test import QUERY_SLOT_LIMIT
 from spectacles.validators.sql import (
     DEFAULT_CHUNK_SIZE,
     DEFAULT_QUERY_CONCURRENCY,
@@ -444,6 +445,7 @@ async def validate_data_tests(
         self,
         ref: Optional[str] = None,
         filters: Optional[List[str]] = None,
+        query_slot_limit: int = QUERY_SLOT_LIMIT,
     ) -> JsonDict:
         if filters is None:
             filters = ["*/*"]
@@ -462,7 +464,7 @@ async def validate_data_tests(
                 f"{'explore' if explore_count == 1 else 'explores'}"
             )
             tests = await validator.get_tests(project)
-            await validator.validate(tests)
+            await validator.validate(tests, query_slot_limit)
 
         results = project.get_results(validator="data_test")
         return results

spectacles/validators/data_test.py

@@ -94,9 +94,11 @@ async def get_tests(self, project: Project) -> List[DataTest]:
 
         return selected_tests
 
-    async def validate(self, tests: List[DataTest]) -> List[DataTestError]:
+    async def validate(
+        self, tests: List[DataTest], query_slot_limit: int = QUERY_SLOT_LIMIT
+    ) -> List[DataTestError]:
         data_test_errors: List[DataTestError] = []
-        query_slot = asyncio.Semaphore(QUERY_SLOT_LIMIT)
+        query_slot = asyncio.Semaphore(query_slot_limit)
 
         async def run_test(test: DataTest, query_slot: asyncio.Semaphore) -> None:
             async with query_slot: