fix: added run_multiprocess

Author: Dmitriy Astapkovich

examples/example_app.py

@@ -89,20 +89,15 @@ def run(self) -> None:
         asyncio.run(self._test_async_runtime())
 
 
-if __name__ == "__main__":
-    app.add_worker(
-        "worker_1",
-        "task_q_1",
-        activities=[test_boost_activity_1, test_boost_activity_3],
-    )
-    app.add_worker("worker_2", "task_q_2", activities=[test_boost_activity_2])
-
-    boost_worker = app.add_worker("worker_3", "task_q_3", workflows=[MyWorkflow])
-    boost_worker.configur_temporal_client(use_pydantic_data_converter=True)
-    boost_worker.configur_temporal_runtime(prometheus_bind_address="0.0.0.0:8801")
-
-    app.add_asgi_worker("asgi_worker", fastapi_app, "0.0.0.0", 8001)
-
-    app.add_exec_method_sync("migrate_db", fake_db_migration)
+app.add_worker(
+    "worker_1",
+    "task_q_1",
+    activities=[test_boost_activity_1, test_boost_activity_3],
+)
+app.add_worker("worker_2", "task_q_2", activities=[test_boost_activity_2])
+boost_worker = app.add_worker("worker_3", "task_q_3", workflows=[MyWorkflow])
+boost_worker.configure_temporal_client(use_pydantic_data_converter=True)
+boost_worker.configur_temporal_runtime(prometheus_bind_address="0.0.0.0:8801")
 
-    app.run()
+app.add_asgi_worker("asgi_worker", fastapi_app, "0.0.0.0", 8001)
+app.add_exec_method_sync("migrate_db", fake_db_migration)

pyproject.toml

@@ -1,36 +1,36 @@
 [tool.poetry]
-name = "temporal-boost"
-version = "1.2.0"
-description = "Small framework for Temporal development"
 authors = [
-    "northpowered <[email protected]>",
-    "dmastapkovich <[email protected]>",
+  "northpowered <[email protected]>",
+  "dmastapkovich <[email protected]>",
 ]
+description = "Small framework for Temporal development"
+keywords = ["temporal", "framework", "development", "python"]
 license = "MIT"
-readme = "README.md"
+name = "temporal-boost"
 packages = [{include = "temporal_boost"}]
+readme = "README.md"
 repository = "https://github.com/northpowered/temporal-boost"
-keywords = ["temporal", "framework", "development", "python"]
+version = "1.2.0"
 
 classifiers = [
-    "Programming Language :: Python :: 3",
-    "Programming Language :: Python :: 3.10",
-    "Programming Language :: Python :: 3.11",
-    "Programming Language :: Python :: 3.12",
-    "Programming Language :: Python :: 3.13",
-    "License :: OSI Approved :: MIT License",
-    "Operating System :: OS Independent",
+  "Programming Language :: Python :: 3",
+  "Programming Language :: Python :: 3.10",
+  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
+  "License :: OSI Approved :: MIT License",
+  "Operating System :: OS Independent",
 ]
 
 [tool.poetry.dependencies]
-python = "^3.10.0"
-typer = "^0.15.0"
-temporalio = "^1.10.0"
+granian = {version = "^2.0.0", optional = true}
+hypercorn = {version = "^0.17.0", optional = true}
 pydantic = "^2.10.0"
+python = "^3.10.0"
 pyyaml = "^6.0.0"
+temporalio = "^1.10.0"
+typer = "^0.15.0"
 uvicorn = {version = "^0.30.0", optional = true}
-granian = {version = "^2.0.0", optional = true}
-hypercorn = {version = "^0.17.0", optional = true}
 uvloop = {version = "^0.21.0", optional = true}
 
 [tool.poetry.extras]
@@ -40,6 +40,9 @@ hypercorn = ["hypercorn"]
 uvicorn = ["uvicorn"]
 uvloop = ["uvloop"]
 
+[tool.poetry.scripts]
+temporal-boost = "temporal_boost.__main__:cli"
+
 [tool.poetry.group.dev.dependencies]
 fastapi = "^0.110.0"
 pre-commit = "^3.8.0"

temporal_boost/__main__.py

@@ -0,0 +1,5 @@
+from temporal_boost.cli.runner import cli
+
+
+if __name__ == "__main__":
+    cli()

temporal_boost/boost_app.py

@@ -39,7 +39,7 @@ def __init__(  # noqa: PLR0913
         self._name: str = name or "temporal_generic_service"
         self._loop_impl = loop_impl
 
-        self._global_temporal_endpointe = temporal_endpoint
+        self._global_temporal_endpoint = temporal_endpoint
         self._global_temporal_namespace = temporal_namespace
         self._global_use_pydantic = use_pydantic
 
@@ -88,13 +88,13 @@ def add_worker(  # noqa: PLR0913
             debug_mode=self._debug_mode,
             **worker_kwargs,
         )
-        worker.configur_temporal_client(
-            target_host=self._global_temporal_endpointe,
+        worker.configure_temporal_client(
+            target_host=self._global_temporal_endpoint,
             namespace=self._global_temporal_namespace,
         )
 
         if self._global_use_pydantic is not None:
-            worker.configur_temporal_client(use_pydantic_data_converter=self._global_use_pydantic)
+            worker.configure_temporal_client(use_pydantic_data_converter=self._global_use_pydantic)
 
         self._run_typer.command(name=worker_name)(worker.run)
 
@@ -147,9 +147,9 @@ def add_async_runtime(self, worker_name: str, boost_worker: TemporalBoostWorker)
         self._registered_workers.append(boost_worker)
         logger.info(f"Async runtime {worker_name} was registered in CLI")
 
-    def run(self) -> None:
+    def run(self, *args: Any, **kwargs: Any) -> None:
         self._loop = loops.get(self._loop_impl)
-        self._loop.run_until_complete(self._root_typer())
+        self._loop.run_until_complete(self._root_typer(*args, **kwargs))
 
     @staticmethod
     def _configure_logging(log_config: dict[str, Any] | Path | None) -> None:

temporal_boost/cli/__init__.py

@@ -0,0 +1,58 @@
+import importlib
+import logging
+from multiprocessing import Process
+from typing import TYPE_CHECKING, Any
+
+import typer
+
+
+if TYPE_CHECKING:
+    from types import ModuleType
+
+
+logger = logging.getLogger(__name__)
+cli = typer.Typer(help="CLI runner for BoostApp services")
+
+
+def _import_app_object(app_path: str) -> Any:
+    if ":" not in app_path:
+        raise typer.BadParameter("App path must be in format 'module.submodule:app'")
+    module_path, app_attr = app_path.split(":")
+    module: ModuleType = importlib.import_module(module_path)
+    app = getattr(module, app_attr, None)
+    if app is None:
+        raise typer.BadParameter(f"No '{app_attr}' found in module '{module_path}'")
+    return app
+
+
+def _run_single(app_path: str, run_args: list[str]) -> None:
+    app = _import_app_object(app_path)
+    app.run(run_args)
+
+
+def _run_multiprocess(app_path: str, processes: int, run_args: list[str]) -> None:
+    process_list: list[Process] = []
+    for index in range(processes):
+        process = Process(target=_run_single, args=(app_path, run_args), name=f"worker-{index}")
+        process_list.append(process)
+        process.start()
+    for process in process_list:
+        process.join()
+
+
+@cli.command("run")
+def run(
+    app: str = typer.Argument(..., help="Path to BoostApp, e.g. 'my_app.app:app'"),
+    workers: int = typer.Option(1, "--workers", "-w", help="Number of processes to run"),
+    run_args: list[str] = typer.Argument(None, help="Additional arguments to pass to app.run()", show_default=False),  # noqa: B008
+) -> None:
+    additional_args = run_args or []
+    if workers > 1:
+        logger.info(f"Starting {workers} processes for app '{app}' with arguments: {additional_args}")
+        _run_multiprocess(app, workers, additional_args)
+    else:
+        _run_single(app, additional_args)
+
+
+if __name__ == "__main__":
+    cli()

temporal_boost/cli/runner.py

@@ -1,3 +1,5 @@
+import logging
+import socket
 from collections.abc import Mapping
 
 from temporalio.runtime import (
@@ -12,6 +14,16 @@
 from temporal_boost.temporal import config
 
 
+logger = logging.getLogger(__name__)
+
+
+def is_port_in_use(host: str, port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.settimeout(1.0)
+        result = sock.connect_ex((host, port))
+        return result == 0
+
+
 def create_runtime(  # noqa: PLR0913
     *,
     logging: LoggingConfig | None = None,
@@ -30,17 +42,42 @@ def create_runtime(  # noqa: PLR0913
         global_tags = {}
 
     if metrics is None and prometheus_bind_address is not None:
-        metrics = PrometheusConfig(
-            bind_address=prometheus_bind_address,
-            counters_total_suffix=prometheus_counters_total_suffix or False,
-            unit_suffix=prometheus_unit_suffix or False,
-            durations_as_seconds=prometheus_durations_as_seconds or False,
-        )
+        try:
+            host, port_str = prometheus_bind_address.split(":")
+            port = int(port_str)
+        except Exception as exc:
+            raise ValueError("Invalid prometheus_bind_address format, expected 'host:port'") from exc
+
+        if is_port_in_use(host, port):
+            logger.warning(f"Port {port} on {host} is already in use. Disabling Prometheus metrics.")
+            metrics = None
+        else:
+            metrics = PrometheusConfig(
+                bind_address=prometheus_bind_address,
+                counters_total_suffix=prometheus_counters_total_suffix or False,
+                unit_suffix=prometheus_unit_suffix or False,
+                durations_as_seconds=prometheus_durations_as_seconds or False,
+            )
+
     telemetry_config = TelemetryConfig(
         logging=logging,
         metrics=metrics,
         global_tags=global_tags,
         attach_service_name=attach_service_name,
         metric_prefix=metric_prefix,
     )
-    return Runtime(telemetry=telemetry_config)
+    try:
+        return Runtime(telemetry=telemetry_config)
+    except ValueError as err:
+        err_msg = str(err)
+        if "Address already in use" in err_msg:
+            logger.warning("Prometheus exporter port is already in use. Disabling metrics for this process.")
+            telemetry_config_no_metrics = TelemetryConfig(
+                logging=logging,
+                metrics=None,
+                global_tags=global_tags,
+                attach_service_name=attach_service_name,
+                metric_prefix=metric_prefix,
+            )
+            return Runtime(telemetry=telemetry_config_no_metrics)
+        raise

temporal_boost/temporal/runtime.py

@@ -70,7 +70,7 @@ def __init__(  # noqa: PLR0913
     def temporal_client(self) -> Client:
         if not self._client:
             raise RuntimeError(
-                "Temporal client has not been initialized. Ensure 'configur_temporal_client()' has been called"
+                "Temporal client has not been initialized. Ensure 'configure_temporal_client()' has been called"
             )
         return self._client
 
@@ -92,7 +92,7 @@ def temporal_client_runtime(self) -> Runtime:
             self.configur_temporal_runtime()
         return cast("Runtime", self._runtime)
 
-    def configur_temporal_client(  # noqa: PLR0913
+    def configure_temporal_client(  # noqa: PLR0913
         self,
         *,
         target_host: str | None = None,
@@ -161,7 +161,7 @@ def configur_temporal_runtime(  # noqa: PLR0913
 
     async def _build_worker(self) -> None:
         if not self._client_builder:
-            self.configur_temporal_client()
+            self.configure_temporal_client()
             self._client_builder = cast("TemporalClientBuilder", self._client_builder)
 
         self._client = await self._client_builder.build()