def _on_code_change_wrapper()

in samcli/lib/sync/watch_manager.py [0:0]

• 20 lines of code
• 12 McCabe index (conditional complexity)

    def _on_code_change_wrapper(self, resource_id: ResourceIdentifier) -> OnChangeCallback:
        """Wrapper method that generates a callback for code changes.

        Parameters
        ----------
        resource_id : ResourceIdentifier
            Resource that associates to the callback

        Returns
        -------
        OnChangeCallback
            Callback function
        """

        def on_code_change(event: Optional[FileSystemEvent] = None) -> None:
            """
            Custom event handling to create a new sync flow if a file was modified.

            Parameters
            ----------
            event: Optional[FileSystemEvent]
                The event that triggered the change
            """
            if event and event.event_type == EVENT_TYPE_OPENED:
                # Ignore all file opened events since this event is
                # added in addition to a create or modified event,
                # causing an infinite loop of sync flow creations
                LOG.debug("Ignoring file system OPENED event")
                return

            if (
                platform.system().lower() == "linux"
                and event
                and event.event_type == EVENT_TYPE_MODIFIED
                and event.is_directory
            ):
                # Linux machines appear to emit an additional event when
                # a file gets updated; a folder modfied event
                # If folder/file.txt gets updated, there will be two events:
                #   1. file.txt modified event
                #   2. folder modified event
                # We want to ignore the second event
                #
                # It looks like the other way a folder modified event can happen
                # is if the permissions of the folder were changed
                LOG.debug(f"Ignoring file system MODIFIED event for folder {event.src_path!r}")
                return

            # sync flow factory should always exist, but guarding just incase
            if not self._sync_flow_factory:
                LOG.debug("Sync flow factory not defined, skipping trigger")
                return

            sync_flow = self._sync_flow_factory.create_sync_flow(resource_id)
            if sync_flow and not self._waiting_infra_sync:
                self._sync_flow_executor.add_delayed_sync_flow(sync_flow, dedup=True, wait_time=DEFAULT_WAIT_TIME)

        return on_code_change