|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Custom linter to check for httpx.Client() usage. |
| 3 | +
|
| 4 | +This script checks for direct usage of httpx.Client() without using the |
| 5 | +get_httpx_client_kwargs() function, which is required for proper SSL |
| 6 | +and proxy configuration in the UiPath Python SDK. |
| 7 | +""" |
| 8 | + |
| 9 | +import ast |
| 10 | +import sys |
| 11 | +from pathlib import Path |
| 12 | +from typing import List, NamedTuple |
| 13 | + |
| 14 | + |
| 15 | +class LintViolation(NamedTuple): |
| 16 | + """Represents a linting violation.""" |
| 17 | + |
| 18 | + filename: str |
| 19 | + line: int |
| 20 | + column: int |
| 21 | + message: str |
| 22 | + rule_code: str |
| 23 | + |
| 24 | + |
| 25 | +class HttpxClientChecker(ast.NodeVisitor): |
| 26 | + """AST visitor to check for httpx.Client() usage violations.""" |
| 27 | + |
| 28 | + def __init__(self, filename: str): |
| 29 | + """Initialize the checker with a filename. |
| 30 | +
|
| 31 | + Args: |
| 32 | + filename: The path to the file being checked. |
| 33 | + """ |
| 34 | + self.filename = filename |
| 35 | + self.violations: List[LintViolation] = [] |
| 36 | + self.has_httpx_import = False |
| 37 | + self.has_get_httpx_client_kwargs_import = False |
| 38 | + # Track variables that contain get_httpx_client_kwargs |
| 39 | + self.variables_with_httpx_kwargs: set[str] = set() |
| 40 | + |
| 41 | + def visit_Import(self, node: ast.Import) -> None: |
| 42 | + """Check for httpx imports.""" |
| 43 | + for alias in node.names: |
| 44 | + if alias.name == "httpx": |
| 45 | + self.has_httpx_import = True |
| 46 | + self.generic_visit(node) |
| 47 | + |
| 48 | + def visit_ImportFrom(self, node: ast.ImportFrom) -> None: |
| 49 | + """Check for imports from httpx or get_httpx_client_kwargs.""" |
| 50 | + if node.module == "httpx": |
| 51 | + self.has_httpx_import = True |
| 52 | + elif node.module and "get_httpx_client_kwargs" in [ |
| 53 | + alias.name for alias in (node.names or []) |
| 54 | + ]: |
| 55 | + self.has_get_httpx_client_kwargs_import = True |
| 56 | + self.generic_visit(node) |
| 57 | + |
| 58 | + def visit_Assign(self, node: ast.Assign) -> None: |
| 59 | + """Track variable assignments that use get_httpx_client_kwargs.""" |
| 60 | + if len(node.targets) == 1 and isinstance(node.targets[0], ast.Name): |
| 61 | + var_name = node.targets[0].id |
| 62 | + if self._assignment_uses_get_httpx_client_kwargs(node.value): |
| 63 | + self.variables_with_httpx_kwargs.add(var_name) |
| 64 | + self.generic_visit(node) |
| 65 | + |
| 66 | + def _assignment_uses_get_httpx_client_kwargs(self, value_node: ast.AST) -> bool: |
| 67 | + """Check if an assignment value uses get_httpx_client_kwargs.""" |
| 68 | + if isinstance(value_node, ast.Call): |
| 69 | + # Direct call: var = get_httpx_client_kwargs() |
| 70 | + if isinstance(value_node.func, ast.Name): |
| 71 | + if value_node.func.id == "get_httpx_client_kwargs": |
| 72 | + return True |
| 73 | + elif isinstance(value_node.func, ast.Attribute): |
| 74 | + if value_node.func.attr == "get_httpx_client_kwargs": |
| 75 | + return True |
| 76 | + elif isinstance(value_node, ast.Dict): |
| 77 | + # Dictionary that spreads get_httpx_client_kwargs: {..., **get_httpx_client_kwargs()} |
| 78 | + for key in value_node.keys: |
| 79 | + if key is None: # This is a **kwargs expansion |
| 80 | + # Find corresponding value |
| 81 | + idx = value_node.keys.index(key) |
| 82 | + if idx < len(value_node.values): |
| 83 | + spread_value = value_node.values[idx] |
| 84 | + if isinstance(spread_value, ast.Call): |
| 85 | + if isinstance(spread_value.func, ast.Name): |
| 86 | + if spread_value.func.id == "get_httpx_client_kwargs": |
| 87 | + return True |
| 88 | + elif isinstance(spread_value.func, ast.Attribute): |
| 89 | + if spread_value.func.attr == "get_httpx_client_kwargs": |
| 90 | + return True |
| 91 | + elif isinstance(spread_value, ast.Name): |
| 92 | + # Spreading another variable that might contain httpx kwargs |
| 93 | + if spread_value.id in self.variables_with_httpx_kwargs: |
| 94 | + return True |
| 95 | + return False |
| 96 | + |
| 97 | + def visit_Call(self, node: ast.Call) -> None: |
| 98 | + """Check for httpx.Client() and httpx.AsyncClient() calls.""" |
| 99 | + if self._is_httpx_client_call(node): |
| 100 | + # Check if this is a proper usage with get_httpx_client_kwargs |
| 101 | + if not self._is_using_get_httpx_client_kwargs(node): |
| 102 | + client_type = self._get_client_type(node) |
| 103 | + violation = LintViolation( |
| 104 | + filename=self.filename, |
| 105 | + line=node.lineno, |
| 106 | + column=node.col_offset, |
| 107 | + message=f"Use **get_httpx_client_kwargs() with {client_type}() - should be: {client_type}(**get_httpx_client_kwargs())", |
| 108 | + rule_code="UIPATH001", |
| 109 | + ) |
| 110 | + self.violations.append(violation) |
| 111 | + |
| 112 | + self.generic_visit(node) |
| 113 | + |
| 114 | + def _is_httpx_client_call(self, node: ast.Call) -> bool: |
| 115 | + """Check if the call is httpx.Client() or httpx.AsyncClient().""" |
| 116 | + if isinstance(node.func, ast.Attribute): |
| 117 | + if ( |
| 118 | + isinstance(node.func.value, ast.Name) |
| 119 | + and node.func.value.id == "httpx" |
| 120 | + and node.func.attr in ("Client", "AsyncClient") |
| 121 | + ): |
| 122 | + return True |
| 123 | + elif isinstance(node.func, ast.Name) and node.func.id in ( |
| 124 | + "Client", |
| 125 | + "AsyncClient", |
| 126 | + ): |
| 127 | + # This could be a direct Client/AsyncClient import, check if httpx is imported |
| 128 | + return self.has_httpx_import |
| 129 | + return False |
| 130 | + |
| 131 | + def _get_client_type(self, node: ast.Call) -> str: |
| 132 | + """Get the client type name (Client or AsyncClient).""" |
| 133 | + if isinstance(node.func, ast.Attribute): |
| 134 | + return f"httpx.{node.func.attr}" |
| 135 | + elif isinstance(node.func, ast.Name): |
| 136 | + return node.func.id |
| 137 | + return "httpx.Client" |
| 138 | + |
| 139 | + def _is_using_get_httpx_client_kwargs(self, node: ast.Call) -> bool: |
| 140 | + """Check if the httpx.Client() call is using **get_httpx_client_kwargs().""" |
| 141 | + # Check if there are any **kwargs that use get_httpx_client_kwargs directly |
| 142 | + for keyword in node.keywords: |
| 143 | + if keyword.arg is None: # This is a **kwargs expansion |
| 144 | + if isinstance(keyword.value, ast.Call): |
| 145 | + if isinstance(keyword.value.func, ast.Name): |
| 146 | + if keyword.value.func.id == "get_httpx_client_kwargs": |
| 147 | + return True |
| 148 | + elif isinstance(keyword.value.func, ast.Attribute): |
| 149 | + if keyword.value.func.attr == "get_httpx_client_kwargs": |
| 150 | + return True |
| 151 | + elif isinstance(keyword.value, ast.Name): |
| 152 | + # Check if this variable might contain get_httpx_client_kwargs |
| 153 | + # This handles cases like: **client_kwargs where client_kwargs was built from get_httpx_client_kwargs |
| 154 | + var_name = keyword.value.id |
| 155 | + if self._variable_contains_get_httpx_client_kwargs(var_name): |
| 156 | + return True |
| 157 | + |
| 158 | + # Also check if it's the ONLY argument and it's **get_httpx_client_kwargs() |
| 159 | + # This handles cases like: httpx.Client(**get_httpx_client_kwargs()) |
| 160 | + if len(node.args) == 0 and len(node.keywords) == 1: |
| 161 | + keyword = node.keywords[0] |
| 162 | + if keyword.arg is None and isinstance(keyword.value, ast.Call): |
| 163 | + if isinstance(keyword.value.func, ast.Name): |
| 164 | + if keyword.value.func.id == "get_httpx_client_kwargs": |
| 165 | + return True |
| 166 | + elif isinstance(keyword.value.func, ast.Attribute): |
| 167 | + if keyword.value.func.attr == "get_httpx_client_kwargs": |
| 168 | + return True |
| 169 | + |
| 170 | + return False |
| 171 | + |
| 172 | + def _variable_contains_get_httpx_client_kwargs(self, var_name: str) -> bool: |
| 173 | + """Check if a variable was built using get_httpx_client_kwargs().""" |
| 174 | + # Check if we've tracked this variable as containing httpx kwargs |
| 175 | + if var_name in self.variables_with_httpx_kwargs: |
| 176 | + return True |
| 177 | + |
| 178 | + # Fallback: Simple heuristic based on naming patterns |
| 179 | + # This handles cases where the variable assignment might be complex |
| 180 | + if "client_kwargs" in var_name.lower() or "httpx_kwargs" in var_name.lower(): |
| 181 | + return True |
| 182 | + |
| 183 | + return False |
| 184 | + |
| 185 | + |
| 186 | +def check_file(filepath: Path) -> List[LintViolation]: |
| 187 | + """Check a single Python file for httpx.Client() violations.""" |
| 188 | + try: |
| 189 | + with open(filepath, "r", encoding="utf-8") as f: |
| 190 | + content = f.read() |
| 191 | + |
| 192 | + tree = ast.parse(content, filename=str(filepath)) |
| 193 | + checker = HttpxClientChecker(str(filepath)) |
| 194 | + checker.visit(tree) |
| 195 | + return checker.violations |
| 196 | + |
| 197 | + except SyntaxError: |
| 198 | + # Skip files with syntax errors |
| 199 | + return [] |
| 200 | + except Exception as e: |
| 201 | + print(f"Error checking {filepath}: {e}", file=sys.stderr) |
| 202 | + return [] |
| 203 | + |
| 204 | + |
| 205 | +def main(): |
| 206 | + """Main function to run the linter.""" |
| 207 | + if len(sys.argv) > 1: |
| 208 | + paths = [Path(p) for p in sys.argv[1:]] |
| 209 | + else: |
| 210 | + # Default to checking src and tests directories |
| 211 | + paths = [Path("src"), Path("tests")] |
| 212 | + |
| 213 | + all_violations = [] |
| 214 | + |
| 215 | + for path in paths: |
| 216 | + if path.is_file() and path.suffix == ".py": |
| 217 | + violations = check_file(path) |
| 218 | + all_violations.extend(violations) |
| 219 | + elif path.is_dir(): |
| 220 | + for py_file in path.rglob("*.py"): |
| 221 | + violations = check_file(py_file) |
| 222 | + all_violations.extend(violations) |
| 223 | + |
| 224 | + # Report violations |
| 225 | + if all_violations: |
| 226 | + for violation in all_violations: |
| 227 | + print( |
| 228 | + f"{violation.filename}:{violation.line}:{violation.column}: {violation.rule_code} {violation.message}" |
| 229 | + ) |
| 230 | + sys.exit(1) |
| 231 | + else: |
| 232 | + print("No httpx.Client() violations found.") |
| 233 | + sys.exit(0) |
| 234 | + |
| 235 | + |
| 236 | +if __name__ == "__main__": |
| 237 | + main() |
0 commit comments