Merge pull request #106 from opsmill/pog-ruf-ruleset

Correct autofixable ruff violations

Author: ogenstad

infrahub_sdk/_importer.py

@@ -27,7 +27,7 @@ def import_module(
     try:
         module = importlib.import_module(module_name)
     except ModuleNotFoundError as exc:
-        raise ModuleImportError(message=f"{str(exc)} ({module_path})") from exc
+        raise ModuleImportError(message=f"{exc!s} ({module_path})") from exc
     except SyntaxError as exc:
         raise ModuleImportError(message=str(exc)) from exc
 

infrahub_sdk/ctl/check.py

@@ -123,7 +123,7 @@ async def run_check(
                     log.error(f"  {log_message['message']}")
 
     except QueryNotFoundError as exc:
-        log.warning(f"{module_name}::{check}: unable to find query ({str(exc)})")
+        log.warning(f"{module_name}::{check}: unable to find query ({exc!s})")
         passed = False
     except Exception as exc:  # pylint: disable=broad-exception-caught
         log.warning(f"{module_name}::{check}: An error occurred during execution ({exc})")

infrahub_sdk/ctl/cli_commands.py

@@ -366,7 +366,7 @@ def transform(
 
 @app.command(name="protocols")
 @catch_exception(console=console)
-def protocols(  # noqa: PLR0915
+def protocols(
     schemas: list[Path] = typer.Option(None, help="List of schemas or directory to load."),
     branch: str = typer.Option(None, help="Branch of schema to export Python protocols for."),
     sync: bool = typer.Option(False, help="Generate for sync or async."),

infrahub_sdk/ctl/utils.py

@@ -48,38 +48,38 @@ def handle_exception(exc: Exception, console: Console, exit_code: int) -> NoRetu
     if isinstance(exc, Exit):
         raise typer.Exit(code=exc.exit_code)
     if isinstance(exc, AuthenticationError):
-        console.print(f"[red]Authentication failure: {str(exc)}")
+        console.print(f"[red]Authentication failure: {exc!s}")
         raise typer.Exit(code=exit_code)
     if isinstance(exc, (ServerNotReachableError, ServerNotResponsiveError)):
-        console.print(f"[red]{str(exc)}")
+        console.print(f"[red]{exc!s}")
         raise typer.Exit(code=exit_code)
     if isinstance(exc, HTTPError):
-        console.print(f"[red]HTTP communication failure: {str(exc)} on {exc.request.method} to {exc.request.url}")
+        console.print(f"[red]HTTP communication failure: {exc!s} on {exc.request.method} to {exc.request.url}")
         raise typer.Exit(code=exit_code)
     if isinstance(exc, GraphQLError):
         print_graphql_errors(console=console, errors=exc.errors)
         raise typer.Exit(code=exit_code)
     if isinstance(exc, (SchemaNotFoundError, NodeNotFoundError)):
-        console.print(f"[red]Error: {str(exc)}")
+        console.print(f"[red]Error: {exc!s}")
         raise typer.Exit(code=exit_code)
 
-    console.print(f"[red]Error: {str(exc)}")
+    console.print(f"[red]Error: {exc!s}")
     console.print(traceback.format_exc())
     raise typer.Exit(code=exit_code)
 
 
 def catch_exception(
     console: Optional[Console] = None, exit_code: int = 1
-) -> Callable[[Callable[..., T]], Callable[..., Union[T, Coroutine[Any, Any, T], NoReturn]]]:
+) -> Callable[[Callable[..., T]], Callable[..., Union[T, Coroutine[Any, Any, T]]]]:
     """Decorator to handle exception for commands."""
     if not console:
         console = Console()
 
-    def decorator(func: Callable[..., T]) -> Callable[..., Union[T, Coroutine[Any, Any, T], NoReturn]]:
+    def decorator(func: Callable[..., T]) -> Callable[..., Union[T, Coroutine[Any, Any, T]]]:
         if asyncio.iscoroutinefunction(func):
 
             @wraps(func)
-            async def async_wrapper(*args: Any, **kwargs: Any) -> Union[T, NoReturn]:
+            async def async_wrapper(*args: Any, **kwargs: Any) -> T:
                 try:
                     return await func(*args, **kwargs)
                 except (Error, Exception) as exc:  # pylint: disable=broad-exception-caught
@@ -88,7 +88,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Union[T, NoReturn]:
             return async_wrapper
 
         @wraps(func)
-        def wrapper(*args: Any, **kwargs: Any) -> Union[T, NoReturn]:
+        def wrapper(*args: Any, **kwargs: Any) -> T:
             try:
                 return func(*args, **kwargs)
             except (Error, Exception) as exc:  # pylint: disable=broad-exception-caught

infrahub_sdk/graphql.py

@@ -48,12 +48,12 @@ def render_query_block(data: dict, offset: int = 4, indentation: int = 4) -> lis
         elif isinstance(value, dict) and len(value) == 1 and ALIAS_KEY in value and value[ALIAS_KEY]:
             lines.append(f"{offset_str}{value[ALIAS_KEY]}: {key}")
         elif isinstance(value, dict):
-            if ALIAS_KEY in value and value[ALIAS_KEY]:
+            if value.get(ALIAS_KEY):
                 key_str = f"{value[ALIAS_KEY]}: {key}"
             else:
                 key_str = key
 
-            if FILTERS_KEY in value and value[FILTERS_KEY]:
+            if value.get(FILTERS_KEY):
                 filters_str = ", ".join(
                     [f"{key2}: {convert_to_graphql_as_string(value2)}" for key2, value2 in value[FILTERS_KEY].items()]
                 )

infrahub_sdk/pytest_plugin/items/__init__.py

@@ -13,12 +13,12 @@
 )
 
 __all__ = [
-    "InfrahubItem",
     "InfrahubCheckIntegrationItem",
     "InfrahubCheckSmokeItem",
     "InfrahubCheckUnitProcessItem",
     "InfrahubGraphQLQueryIntegrationItem",
     "InfrahubGraphQLQuerySmokeItem",
+    "InfrahubItem",
     "InfrahubJinja2TransformIntegrationItem",
     "InfrahubJinja2TransformSmokeItem",
     "InfrahubJinja2TransformUnitRenderItem",

infrahub_sdk/query_groups.py

@@ -49,7 +49,7 @@ def _get_params_as_str(self) -> str:
         """Convert the params in dict format, into a string"""
         params_as_str: list[str] = []
         for key, value in self.params.items():
-            params_as_str.append(f"{key}: {str(value)}")
+            params_as_str.append(f"{key}: {value!s}")
         return ", ".join(params_as_str)
 
     def _generate_group_name(self, suffix: Optional[str] = None) -> str:

pyproject.toml

@@ -250,13 +250,8 @@ ignore = [
     "PTH109",  # `os.getcwd()` should be replaced by `Path.cwd()`
     "RET504",  # Unnecessary assignment to `data` before `return` statement
     "RUF005",  # Consider `[*path, str(key)]` instead of concatenation
-    "RUF010",  # [*] Use explicit conversion flag
     "RUF015",  # Prefer `next(iter(input_data["variables"].keys()))` over single element slice
-    "RUF019",  # [*] Unnecessary key check before dictionary access
-    "RUF020",  # [*] `Union[NoReturn, T]` is equivalent to `T`
-    "RUF022",  # [*] `__all__` is not sorted
     "RUF029",  # Function is declared `async`, but doesn't `await` or use `async` features.
-    "RUF100",  # [*] Unused `noqa` directive
     "S108",    # Probable insecure usage of temporary file or directory
     "S311",    # Standard pseudo-random generators are not suitable for cryptographic purposes
     "S701",    # By default, jinja2 sets `autoescape` to `False`. Consider using `autoescape=True`

tests/integration/test_export_import.py

@@ -409,7 +409,7 @@ def schema(self, schema_car_base, schema_pool_base, schema_manufacturer_base) ->
         }
 
     @pytest.fixture(scope="class")
-    async def initial_dataset(self, client: InfrahubClient, schema):  # noqa: PLR0914
+    async def initial_dataset(self, client: InfrahubClient, schema):
         await client.schema.load(schemas=[schema])
 
         bmw = await client.create(

tests/unit/sdk/test_schema.py

@@ -310,5 +310,5 @@ async def test_display_schema_load_errors_details_namespace(mock_get_node):
         output = console.file.getvalue()
         expected_console = """Unable to load the schema:
   Node: OuTInstance | namespace (OuT) | String should match pattern '^[A-Z]+$' (string_pattern_mismatch)
-"""  # noqa: E501
+"""
         assert output == expected_console