Method _detect_websocket_usage

zappa/cli.py:3441–3497

Walk the project directory looking for zappa.websocket imports. Returns the dotted module path of the first file found (e.g. ``"ws_handlers"`` or ``"mypackage.ws"``), or ``None`` if no WebSocket usage is detected. The return value is truthy/falsy so existing ``if se

()

    @staticmethod
    def _detect_websocket_usage():
        """Walk the project directory looking for zappa.websocket imports.

        Returns the dotted module path of the first file found (e.g.
        ``"ws_handlers"`` or ``"mypackage.ws"``), or ``None`` if no
        WebSocket usage is detected.  The return value is truthy/falsy
        so existing ``if self.use_websocket:`` checks still work.
        """
        import ast

        skip_dirs = {".", "__pycache__", "node_modules", ".git", ".tox", ".eggs", "venv", "env", ".venv"}
        for dirpath, dirnames, filenames in os.walk("."):
            # Skip hidden dirs, venvs, caches
            dirnames[:] = [d for d in dirnames if d not in skip_dirs and not d.startswith(".")]
            for fname in filenames:
                if not fname.endswith(".py"):
                    continue
                fpath = os.path.join(dirpath, fname)
                try:
                    with open(fpath, "r", encoding="utf-8", errors="ignore") as f:
                        source = f.read()
                except OSError:
                    continue
                if "zappa.websocket" not in source:
                    continue
                try:
                    tree = ast.parse(source, filename=fpath)
                except SyntaxError:
                    continue
                for node in ast.walk(tree):
                    has_import = False
                    if (
                        isinstance(node, ast.ImportFrom)
                        and node.module
                        and (node.module == "zappa.websocket" or node.module.startswith("zappa.websocket."))
                    ):
                        has_import = True
                    elif isinstance(node, ast.Import):
                        for alias in node.names:
                            if alias.name and (alias.name == "zappa.websocket" or alias.name.startswith("zappa.websocket.")):
                                has_import = True
                                break
                    if has_import:
                        # Convert file path to dotted module name
                        # "./ws_handlers.py" -> "ws_handlers"
                        # "./pkg/ws.py" -> "pkg.ws"
                        # "./pkg/__init__.py" -> "pkg"
                        module_path = os.path.normpath(fpath)
                        if module_path.startswith("." + os.sep):
                            module_path = module_path[2:]
                        module_path = module_path[: -len(".py")]
                        if module_path.endswith(os.sep + "__init__") or module_path == "__init__":
                            module_path = module_path[: -len("__init__")].rstrip(os.sep)
                            if not module_path:
                                continue
                        return module_path.replace(os.sep, ".")
        return None