fix: correct CLI factory behavior (#1838)

* fix (#1837): explicit asyncio.Event usage

* fix: correct CLI object import in factory case

* chore: bump version

* chore: revert anyio

* fix: respect extra reload extensions CLI option

faststream/__about__.py

@@ -1,5 +1,5 @@
 """Simple and fast framework to create message brokers based microservices."""
 
-__version__ = "0.5.25"
+__version__ = "0.5.26"
 
 SERVICE_NAME = f"faststream-{__version__}"

faststream/cli/docs/app.py

@@ -56,7 +56,7 @@ def serve(
         if app_dir:  # pragma: no branch
             sys.path.insert(0, app_dir)
 
-        module, _ = import_from_string(app)
+        module, _ = import_from_string(app, is_factory=is_factory)
 
         module_parent = module.parent
         extra_extensions: Sequence[str] = ()
@@ -121,9 +121,8 @@ def gen(
     if app_dir:  # pragma: no branch
         sys.path.insert(0, app_dir)
 
-    _, app_obj = import_from_string(app)
-    if callable(app_obj) and is_factory:
-        app_obj = app_obj()
+    _, app_obj = import_from_string(app, is_factory=is_factory)
+
     raw_schema = get_app_schema(app_obj)
 
     if yaml:
@@ -155,9 +154,8 @@ def _parse_and_serve(
     is_factory: bool = False,
 ) -> None:
     if ":" in app:
-        _, app_obj = import_from_string(app)
-        if callable(app_obj) and is_factory:
-            app_obj = app_obj()
+        _, app_obj = import_from_string(app, is_factory=is_factory)
+
         raw_schema = get_app_schema(app_obj)
 
     else:

faststream/cli/main.py

@@ -115,7 +115,7 @@ def run(
         sys.path.insert(0, app_dir)
 
     # Should be imported after sys.path changes
-    module_path, app_obj = import_from_string(app)
+    module_path, app_obj = import_from_string(app, is_factory=is_factory)
 
     args = (app, extra, is_factory, casted_log_level)
 
@@ -139,6 +139,7 @@ def run(
                 target=_run,
                 args=args,
                 reload_dirs=reload_dirs,
+                extra_extensions=watch_extensions,
             ).run()
 
     elif workers > 1:
@@ -167,9 +168,7 @@ def _run(
     app_level: int = logging.INFO,
 ) -> None:
     """Runs the specified application."""
-    _, app_obj = import_from_string(app)
-    if is_factory and callable(app_obj):
-        app_obj = app_obj()
+    _, app_obj = import_from_string(app, is_factory=is_factory)
 
     if not isinstance(app_obj, Application):
         raise typer.BadParameter(
@@ -242,9 +241,7 @@ def publish(
         if not message:
             raise ValueError("Message parameter is required.")
 
-        _, app_obj = import_from_string(app)
-        if callable(app_obj) and is_factory:
-            app_obj = app_obj()
+        _, app_obj = import_from_string(app, is_factory=is_factory)
 
         if not app_obj.broker:
             raise ValueError("Broker instance not found in the app.")

faststream/cli/utils/imports.py

@@ -67,7 +67,23 @@ def get_app_path(app: str) -> Tuple[Path, str]:
     return mod_path, app_name
 
 
-def import_from_string(import_str: str) -> Tuple[Path, "FastStream"]:
+def import_from_string(
+    import_str: str,
+    *,
+    is_factory: bool = False,
+) -> Tuple[Path, "FastStream"]:
+    module_path, instance = _import_obj_or_factory(import_str)
+
+    if is_factory:
+        if callable(instance):
+            instance = instance()
+        else:
+            raise typer.BadParameter(f'"{instance}" is not a factory')
+
+    return module_path, instance
+
+
+def _import_obj_or_factory(import_str: str) -> Tuple[Path, "FastStream"]:
     """Import FastStream application from module specified by a string."""
     if not isinstance(import_str, str):
         raise typer.BadParameter("Given value is not of type string")

tests/cli/test_publish.py

@@ -30,7 +30,7 @@ def test_publish_command_with_redis_options(runner):
     mock_app = get_mock_app(RedisBroker, RedisFastProducer)
 
     with patch(
-        "faststream.cli.main.import_from_string",
+        "faststream.cli.utils.imports._import_obj_or_factory",
         return_value=(None, mock_app),
     ):
         result = runner.invoke(
@@ -72,7 +72,10 @@ def test_publish_command_with_confluent_options(runner):
 
     mock_app = get_mock_app(ConfluentBroker, AsyncConfluentFastProducer)
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, mock_app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory",
+        return_value=(None, mock_app),
+    ):
         result = runner.invoke(
             faststream_app,
             [
@@ -102,7 +105,10 @@ def test_publish_command_with_kafka_options(runner):
 
     mock_app = get_mock_app(KafkaBroker, AioKafkaFastProducer)
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, mock_app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory",
+        return_value=(None, mock_app),
+    ):
         result = runner.invoke(
             faststream_app,
             [
@@ -132,7 +138,10 @@ def test_publish_command_with_nats_options(runner):
 
     mock_app = get_mock_app(NatsBroker, NatsFastProducer)
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, mock_app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory",
+        return_value=(None, mock_app),
+    ):
         result = runner.invoke(
             faststream_app,
             [
@@ -166,7 +175,10 @@ def test_publish_command_with_rabbit_options(runner):
 
     mock_app = get_mock_app(RabbitBroker, AioPikaFastProducer)
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, mock_app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory",
+        return_value=(None, mock_app),
+    ):
         result = runner.invoke(
             faststream_app,
             [

tests/cli/test_run.py

@@ -12,7 +12,9 @@ def test_run_as_asgi(runner: CliRunner):
     app = AsgiFastStream()
     app.run = AsyncMock()
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory", return_value=(None, app)
+    ):
         result = runner.invoke(
             faststream_app,
             [
@@ -35,7 +37,9 @@ def test_run_as_asgi_with_workers(runner: CliRunner, workers: int):
     app = AsgiFastStream()
     app.run = AsyncMock()
 
-    with patch("faststream.cli.main.import_from_string", return_value=(None, app)):
+    with patch(
+        "faststream.cli.utils.imports._import_obj_or_factory", return_value=(None, app)
+    ):
         result = runner.invoke(
             faststream_app,
             [
@@ -64,7 +68,8 @@ def test_run_as_asgi_callable(runner: CliRunner):
     app_factory = Mock(return_value=app)
 
     with patch(
-        "faststream.cli.main.import_from_string", return_value=(None, app_factory)
+        "faststream.cli.utils.imports._import_obj_or_factory",
+        return_value=(None, app_factory),
     ):
         result = runner.invoke(
             faststream_app,
@@ -78,7 +83,7 @@ def test_run_as_asgi_callable(runner: CliRunner):
                 "--factory",
             ],
         )
-        app_factory.assert_called_once()
+        app_factory.assert_called()
         app.run.assert_awaited_once_with(
             logging.INFO, {"host": "0.0.0.0", "port": "8000"}
         )