fix(httpx): fix httpx proxy

Author: ionmincu

.github/workflows/lint.yml

@@ -35,4 +35,7 @@ jobs:
 
       - name: Check formatting
         run: uv run ruff format --check .
+      
+      - name: Check httpx.Client() usage
+        run: uv run python scripts/lint_httpx_client.py
 

justfile

@@ -4,6 +4,7 @@ default: lint format
 
 lint:
     ruff check .
+    python scripts/lint_httpx_client.py
 
 format:
     ruff format --check .
@@ -13,3 +14,7 @@ build:
 
 install:
     uv sync --all-extras
+
+# Test the custom linter
+test-lint-httpx:
+    python scripts/test_httpx_linter.py

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "uipath"
-version = "2.0.77"
+version = "2.0.78"
 description = "Python SDK and CLI for UiPath Platform, enabling programmatic interaction with automation services, process management, and deployment tools."
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.10"

scripts/debug_test.py

@@ -0,0 +1,28 @@
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+# Simple test
+test_code = """
+import httpx
+
+def test():
+    client = httpx.Client()
+    return client
+"""
+
+linter_path = Path(__file__).parent / "lint_httpx_client.py"
+
+with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
+    f.write(test_code)
+    f.flush()
+
+    print(f"Testing file: {f.name}")
+    result = subprocess.run(
+        [sys.executable, str(linter_path), f.name], capture_output=True, text=True
+    )
+
+    print(f"Return code: {result.returncode}")
+    print(f"Stdout: {result.stdout}")
+    print(f"Stderr: {result.stderr}")

scripts/fix_httpx_client.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""Auto-fix script for httpx.Client() violations.
+
+This script automatically fixes violations by adding **get_httpx_client_kwargs()
+to httpx.Client() and httpx.AsyncClient() calls.
+"""
+
+import ast
+import sys
+from pathlib import Path
+
+
+class HttpxClientFixer(ast.NodeTransformer):
+    """AST transformer to fix httpx.Client() usage violations."""
+
+    def __init__(self, filename: str):
+        """Initialize the fixer with a filename.
+
+        Args:
+            filename: The path to the file being fixed.
+        """
+        self.filename = filename
+        self.fixes_applied = 0
+        self.needs_import = False
+        self.has_httpx_import = False
+        self.has_get_httpx_client_kwargs_import = False
+
+    def visit_Import(self, node: ast.Import) -> ast.Import:
+        """Check for httpx imports."""
+        for alias in node.names:
+            if alias.name == "httpx":
+                self.has_httpx_import = True
+        return node
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> ast.ImportFrom:
+        """Check for imports from httpx or get_httpx_client_kwargs."""
+        if node.module == "httpx":
+            self.has_httpx_import = True
+        elif node.module and "get_httpx_client_kwargs" in [
+            alias.name for alias in (node.names or [])
+        ]:
+            self.has_get_httpx_client_kwargs_import = True
+        return node
+
+    def visit_Call(self, node: ast.Call) -> ast.Call:
+        """Fix httpx.Client() and httpx.AsyncClient() calls."""
+        if self._is_httpx_client_call(node):
+            if not self._is_using_get_httpx_client_kwargs(node):
+                # Add **get_httpx_client_kwargs() to the call
+                new_keyword = ast.keyword(
+                    arg=None,  # **kwargs
+                    value=ast.Call(
+                        func=ast.Name(id="get_httpx_client_kwargs", ctx=ast.Load()),
+                        args=[],
+                        keywords=[],
+                    ),
+                )
+
+                # Create a new call with the added keyword
+                new_node = ast.Call(
+                    func=node.func,
+                    args=node.args,
+                    keywords=node.keywords + [new_keyword],
+                )
+
+                # Copy location information
+                ast.copy_location(new_node, node)
+
+                self.fixes_applied += 1
+                self.needs_import = True
+
+                return new_node
+
+        return self.generic_visit(node)
+
+    def _is_httpx_client_call(self, node: ast.Call) -> bool:
+        """Check if the call is httpx.Client() or httpx.AsyncClient()."""
+        if isinstance(node.func, ast.Attribute):
+            if (
+                isinstance(node.func.value, ast.Name)
+                and node.func.value.id == "httpx"
+                and node.func.attr in ("Client", "AsyncClient")
+            ):
+                return True
+        elif isinstance(node.func, ast.Name) and node.func.id in (
+            "Client",
+            "AsyncClient",
+        ):
+            return self.has_httpx_import
+        return False
+
+    def _is_using_get_httpx_client_kwargs(self, node: ast.Call) -> bool:
+        """Check if the call already uses **get_httpx_client_kwargs()."""
+        for keyword in node.keywords:
+            if keyword.arg is None and isinstance(keyword.value, ast.Call):
+                if isinstance(keyword.value.func, ast.Name):
+                    if keyword.value.func.id == "get_httpx_client_kwargs":
+                        return True
+                elif isinstance(keyword.value.func, ast.Attribute):
+                    if keyword.value.func.attr == "get_httpx_client_kwargs":
+                        return True
+        return False
+
+
+def fix_file(filepath: Path) -> bool:
+    """Fix a single Python file for httpx.Client() violations."""
+    try:
+        with open(filepath, "r", encoding="utf-8") as f:
+            content = f.read()
+
+        tree = ast.parse(content, filename=str(filepath))
+        fixer = HttpxClientFixer(str(filepath))
+
+        # Transform the AST
+        new_tree = fixer.visit(tree)
+
+        if fixer.fixes_applied > 0:
+            # Add import if needed and not already present
+            if fixer.needs_import and not fixer.has_get_httpx_client_kwargs_import:
+                # Find a good place to add the import
+                import_added = False
+                for i, node in enumerate(new_tree.body):
+                    if (
+                        isinstance(node, ast.ImportFrom)
+                        and node.module
+                        and "uipath" in node.module
+                    ):
+                        # Add to existing uipath import if possible
+                        if any(
+                            alias.name == "get_httpx_client_kwargs"
+                            for alias in (node.names or [])
+                        ):
+                            break  # Already imported
+                        # Add new import after existing uipath imports
+                        new_import = ast.ImportFrom(
+                            module="uipath._utils._ssl_context",
+                            names=[
+                                ast.alias(name="get_httpx_client_kwargs", asname=None)
+                            ],
+                            level=0,
+                        )
+                        new_tree.body.insert(i + 1, new_import)
+                        import_added = True
+                        break
+
+                if not import_added:
+                    # Add at the beginning after other imports
+                    insert_pos = 0
+                    for i, node in enumerate(new_tree.body):
+                        if isinstance(node, (ast.Import, ast.ImportFrom)):
+                            insert_pos = i + 1
+                        else:
+                            break
+
+                    new_import = ast.ImportFrom(
+                        module="uipath._utils._ssl_context",
+                        names=[ast.alias(name="get_httpx_client_kwargs", asname=None)],
+                        level=0,
+                    )
+                    new_tree.body.insert(insert_pos, new_import)
+
+            # Convert back to code
+            import astor
+
+            fixed_content = astor.to_source(new_tree)
+
+            # Write back to file
+            with open(filepath, "w", encoding="utf-8") as f:
+                f.write(fixed_content)
+
+            print(f"Fixed {fixer.fixes_applied} violations in {filepath}")
+            return True
+
+        return False
+
+    except Exception as e:
+        print(f"Error fixing {filepath}: {e}", file=sys.stderr)
+        return False
+
+
+def main():
+    """Main function to run the fixer."""
+    if len(sys.argv) > 1:
+        paths = [Path(p) for p in sys.argv[1:]]
+    else:
+        # Default to checking src and tests directories
+        paths = [Path("src"), Path("tests")]
+
+    total_files_fixed = 0
+
+    for path in paths:
+        if path.is_file() and path.suffix == ".py":
+            if fix_file(path):
+                total_files_fixed += 1
+        elif path.is_dir():
+            for py_file in path.rglob("*.py"):
+                if fix_file(py_file):
+                    total_files_fixed += 1
+
+    if total_files_fixed > 0:
+        print(f"\nFixed {total_files_fixed} files. Run the linter again to verify.")
+    else:
+        print("No files needed fixing.")
+
+
+if __name__ == "__main__":
+    main()