Move the function read_file from the ctl module to the SDK

changelog/+move-read-file.housekeeping.md

@@ -0,0 +1 @@
+Move the function `read_file` from the ctl module to the SDK.

infrahub_sdk/ctl/repository.py

@@ -8,15 +8,14 @@
 from rich.console import Console
 from rich.table import Table
 
-from infrahub_sdk.ctl.client import initialize_client
-
 from ..async_typer import AsyncTyper
-from ..ctl.exceptions import FileNotValidError
-from ..ctl.utils import init_logging
+from ..exceptions import FileNotValidError
 from ..graphql import Mutation, Query
 from ..schema.repository import InfrahubRepositoryConfig
-from ._file import read_file
+from ..utils import read_file
+from .client import initialize_client
 from .parameters import CONFIG_PARAM
+from .utils import init_logging
 
 app = AsyncTyper()
 console = Console()

infrahub_sdk/exceptions.py

@@ -131,3 +131,9 @@ class UninitializedError(Error):
 
 class InvalidResponseError(Error):
     """Raised when an object requires an initialization step before use"""
+
+
+class FileNotValidError(Error):
+    def __init__(self, name: str, message: str = ""):
+        self.message = message or f"Cannot parse '{name}' content."
+        super().__init__(self.message)

infrahub_sdk/utils.py

@@ -17,7 +17,7 @@
 
 from infrahub_sdk.repository import GitRepoManager
 
-from .exceptions import JsonDecodeError
+from .exceptions import FileNotValidError, JsonDecodeError
 
 if TYPE_CHECKING:
     from graphql import GraphQLResolveInfo
@@ -342,6 +342,16 @@ def write_to_file(path: Path, value: Any) -> bool:
     return written is not None
 
 
+def read_file(file_name: Path) -> str:
+    if not file_name.is_file():
+        raise FileNotValidError(name=str(file_name), message=f"{file_name} is not a valid file")
+    try:
+        with Path.open(file_name, encoding="utf-8") as fobj:
+            return fobj.read()
+    except UnicodeDecodeError as exc:
+        raise FileNotValidError(name=str(file_name), message=f"Unable to read {file_name} with utf-8 encoding") from exc
+
+
 def get_user_permissions(data: list[dict]) -> dict:
     groups = {}
     for group in data:

infrahub_sdk/yaml.py

@@ -8,9 +8,8 @@
 from pydantic import BaseModel, Field
 from typing_extensions import Self
 
-from .ctl._file import read_file
-from .ctl.exceptions import FileNotValidError
-from .utils import find_files
+from .exceptions import FileNotValidError
+from .utils import find_files, read_file
 
 
 class InfrahubFileApiVersion(str, Enum):