feat: add --no-reload flag to disable hot reload in local commands

samcli/commands/local/cli_common/invoke_context.py

@@ -102,6 +102,7 @@ def __init__(
         mount_symlinks: Optional[bool] = False,
         no_mem_limit: Optional[bool] = False,
         function_logical_ids: Optional[Tuple[str, ...]] = None,
+        no_reload: bool = False,
     ) -> None:
         """
         Initialize the context
@@ -220,6 +221,7 @@ def __init__(
 
         self._mount_symlinks: Optional[bool] = mount_symlinks
         self._no_mem_limit = no_mem_limit
+        self._no_reload = no_reload
 
         # Note(xinhol): despite self._function_provider and self._stacks are initialized as None
         # they will be assigned with a non-None value in __enter__() and
@@ -573,6 +575,7 @@ def lambda_runtime(self) -> LambdaRuntime:
                     image_builder,
                     mount_symlinks=self._mount_symlinks,
                     no_mem_limit=self._no_mem_limit,
+                    no_reload=self._no_reload,
                 ),
                 ContainersMode.COLD: LambdaRuntime(
                     self._container_manager,

samcli/commands/local/cli_common/options.py

@@ -264,6 +264,13 @@ def warm_containers_common_options(f):
             type=click.STRING,
             multiple=False,
         ),
+        click.option(
+            "--no-reload",
+            is_flag=True,
+            default=False,
+            help="Disable automatic hot reloading of Lambda functions when code changes are detected. "
+            "Useful on Windows where file watching can cause significant latency with antivirus software.",
+        ),
     ]
 
     # Reverse the list to maintain ordering of options in help text printed with --help

samcli/commands/local/start_api/cli.py

@@ -146,6 +146,7 @@ def cli(
     ssl_cert_file,
     ssl_key_file,
     no_memory_limit,
+    no_reload,
 ):
     """
     `sam local start-api` command entry point
@@ -182,6 +183,7 @@ def cli(
         ssl_cert_file,
         ssl_key_file,
         no_memory_limit,
+        no_reload,
     )  # pragma: no cover
 
 
@@ -215,6 +217,7 @@ def do_cli(  # pylint: disable=R0914
     ssl_cert_file,
     ssl_key_file,
     no_mem_limit,
+    no_reload=False,
 ):
     """
     Implementation of the ``cli`` method, just separated out for unit testing purposes
@@ -261,6 +264,7 @@ def do_cli(  # pylint: disable=R0914
             invoke_images=processed_invoke_images,
             add_host=add_host,
             no_mem_limit=no_mem_limit,
+            no_reload=no_reload,
         ) as invoke_context:
             ssl_context = (ssl_cert_file, ssl_key_file) if ssl_cert_file else None
             service = LocalApiService(

samcli/commands/local/start_lambda/cli.py

@@ -108,6 +108,7 @@ def cli(
     skip_prepare_infra,
     terraform_plan_file,
     no_memory_limit,
+    no_reload,
 ):
     """
     `sam local start-lambda` command entry point
@@ -141,6 +142,7 @@ def cli(
         invoke_image,
         hook_name,
         no_memory_limit,
+        no_reload,
     )  # pragma: no cover
 
 
@@ -171,6 +173,7 @@ def do_cli(  # pylint: disable=R0914
     invoke_image,
     hook_name,
     no_mem_limit,
+    no_reload=False,
 ):
     """
     Implementation of the ``cli`` method, just separated out for unit testing purposes
@@ -218,6 +221,7 @@ def do_cli(  # pylint: disable=R0914
             invoke_images=processed_invoke_images,
             function_logical_ids=function_logical_ids,
             no_mem_limit=no_mem_limit,
+            no_reload=no_reload,
         ) as invoke_context:
             service = LocalLambdaService(lambda_invoke_context=invoke_context, port=port, host=host)
             service.start()

samcli/local/lambdafn/runtime.py

@@ -506,13 +506,29 @@ def clean_runtime_containers(self):
             self._durable_execution_emulator_container = None
 
 
+class _NoOpObserver:
+    """A no-op observer that satisfies the observer interface without doing any file watching."""
+
+    def watch(self, *args, **kwargs):
+        pass
+
+    def unwatch(self, *args, **kwargs):
+        pass
+
+    def start(self, *args, **kwargs):
+        pass
+
+    def stop(self, *args, **kwargs):
+        pass
+
+
 class WarmLambdaRuntime(LambdaRuntime):
     """
     This class extends the LambdaRuntime class to add the Warm containers feature. This class handles the
     warm containers life cycle.
     """
 
-    def __init__(self, container_manager, image_builder, observer=None, mount_symlinks=False, no_mem_limit=False):
+    def __init__(self, container_manager, image_builder, observer=None, mount_symlinks=False, no_mem_limit=False, no_reload=False):
         """
         Initialize the Local Lambda runtime
 
@@ -524,12 +540,17 @@ def __init__(self, container_manager, image_builder, observer=None, mount_symlin
             Instance of the LambdaImage class that can create am image
         warm_containers bool
             Determines if the warm containers is enabled or not.
+        no_reload bool
+            If True, skip creating the file observer for hot reloading.
         """
         self._function_configs = {}
         self._containers = {}
         self._container_lock = threading.Lock()  # Thread-safe container creation
 
-        self._observer = observer if observer else LambdaFunctionObserver(self._on_code_change)
+        if no_reload:
+            self._observer = observer if observer else _NoOpObserver()
+        else:
+            self._observer = observer if observer else LambdaFunctionObserver(self._on_code_change)
 
         super().__init__(container_manager, image_builder, mount_symlinks=mount_symlinks, no_mem_limit=no_mem_limit)
 

tests/unit/commands/local/cli_common/test_invoke_context.py

@@ -779,7 +779,7 @@ def test_must_create_runner_using_warm_containers(
             self.assertEqual(result, runner_mock)
 
             WarmLambdaRuntimeMock.assert_called_with(
-                container_manager_mock, image_mock, mount_symlinks=False, no_mem_limit=False
+                container_manager_mock, image_mock, mount_symlinks=False, no_mem_limit=False, no_reload=False
             )
             lambda_image_patch.assert_called_once_with(download_mock, True, True, invoke_images=None)
             LocalLambdaMock.assert_called_with(

tests/unit/commands/local/start_api/test_cli.py

@@ -57,6 +57,7 @@ def setUp(self):
         self.container_host_interface = "127.0.0.1"
         self.invoke_image = ()
         self.no_mem_limit = False
+        self.no_reload = False
 
     @patch("samcli.commands.local.cli_common.invoke_context.InvokeContext")
     @patch("samcli.commands.local.lib.local_api_service.LocalApiService")
@@ -99,6 +100,7 @@ def test_cli_must_setup_context_and_start_service(self, local_api_service_mock,
             add_host=self.add_host,
             invoke_images={},
             no_mem_limit=self.no_mem_limit,
+            no_reload=self.no_reload,
         )
 
         local_api_service_mock.assert_called_with(
@@ -200,6 +202,19 @@ def test_must_raise_user_exception_on_invalid_imageuri(self, invoke_context_mock
         expected = "invalid imageuri"
         self.assertEqual(msg, expected)
 
+    @patch("samcli.commands.local.cli_common.invoke_context.InvokeContext")
+    @patch("samcli.commands.local.lib.local_api_service.LocalApiService")
+    def test_no_reload_flag_passed_to_invoke_context(self, local_api_service_mock, invoke_context_mock):
+        context_mock = Mock()
+        invoke_context_mock.return_value.__enter__.return_value = context_mock
+        local_api_service_mock.return_value = Mock()
+
+        self.no_reload = True
+        self.call_cli()
+
+        _, kwargs = invoke_context_mock.call_args
+        self.assertTrue(kwargs["no_reload"])
+
     def call_cli(self):
         start_api_cli(
             ctx=self.ctx_mock,
@@ -231,4 +246,5 @@ def call_cli(self):
             disable_authorizer=self.disable_authorizer,
             add_host=self.add_host,
             no_mem_limit=self.no_mem_limit,
+            no_reload=self.no_reload,
         )

tests/unit/commands/local/start_lambda/test_cli.py

@@ -46,6 +46,7 @@ def setUp(self):
         self.invoke_image = ()
         self.hook_name = None
         self.no_mem_limit = False
+        self.no_reload = False
 
     @patch("samcli.commands.local.cli_common.invoke_context.InvokeContext")
     @patch("samcli.commands.local.lib.local_lambda_service.LocalLambdaService")
@@ -87,6 +88,7 @@ def test_cli_must_setup_context_and_start_service(self, local_lambda_service_moc
             invoke_images={},
             function_logical_ids=(),
             no_mem_limit=self.no_mem_limit,
+            no_reload=self.no_reload,
         )
 
         local_lambda_service_mock.assert_called_with(lambda_invoke_context=context_mock, port=self.port, host=self.host)
@@ -162,6 +164,19 @@ def test_must_raise_user_exception_on_no_free_ports(
         msg = str(context.exception)
         self.assertEqual(msg, expected_exception_message)
 
+    @patch("samcli.commands.local.cli_common.invoke_context.InvokeContext")
+    @patch("samcli.commands.local.lib.local_lambda_service.LocalLambdaService")
+    def test_no_reload_flag_passed_to_invoke_context(self, local_lambda_service_mock, invoke_context_mock):
+        context_mock = Mock()
+        invoke_context_mock.return_value.__enter__.return_value = context_mock
+        local_lambda_service_mock.return_value = Mock()
+
+        self.no_reload = True
+        self.call_cli()
+
+        _, kwargs = invoke_context_mock.call_args
+        self.assertTrue(kwargs["no_reload"])
+
     def call_cli(self):
         start_lambda_cli(
             ctx=self.ctx_mock,
@@ -190,4 +205,5 @@ def call_cli(self):
             invoke_image=self.invoke_image,
             hook_name=self.hook_name,
             no_mem_limit=self.no_mem_limit,
+            no_reload=self.no_reload,
         )

tests/unit/local/lambdafn/test_runtime.py

@@ -2346,3 +2346,36 @@ def test_clean_runtime_containers_stops_emulator_container(self, log_mock):
         emulator_container.stop.assert_called_once()
         log_mock.debug.assert_called_with("Stopping durable functions emulator container")
         self.assertIsNone(runtime._durable_execution_emulator_container)
+
+
+class TestWarmLambdaRuntime_no_reload(TestCase):
+    def setUp(self):
+        self.manager_mock = Mock()
+        self.lambda_image_mock = Mock()
+
+    def test_no_reload_false_creates_lambda_function_observer(self):
+        with patch("samcli.local.lambdafn.runtime.LambdaFunctionObserver") as observer_cls_mock:
+            observer_instance = Mock()
+            observer_cls_mock.return_value = observer_instance
+            runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, no_reload=False)
+            observer_cls_mock.assert_called_once()
+            self.assertEqual(runtime._observer, observer_instance)
+
+    def test_no_reload_true_uses_noop_observer(self):
+        with patch("samcli.local.lambdafn.runtime.LambdaFunctionObserver") as observer_cls_mock:
+            runtime = WarmLambdaRuntime(self.manager_mock, self.lambda_image_mock, no_reload=True)
+            observer_cls_mock.assert_not_called()
+            # _NoOpObserver methods should be no-ops
+            runtime._observer.watch(Mock())
+            runtime._observer.unwatch(Mock())
+            runtime._observer.start()
+            runtime._observer.stop()
+
+    def test_no_reload_true_explicit_observer_takes_precedence(self):
+        explicit_observer = Mock()
+        with patch("samcli.local.lambdafn.runtime.LambdaFunctionObserver") as observer_cls_mock:
+            runtime = WarmLambdaRuntime(
+                self.manager_mock, self.lambda_image_mock, observer=explicit_observer, no_reload=True
+            )
+            observer_cls_mock.assert_not_called()
+            self.assertEqual(runtime._observer, explicit_observer)