commit

debug/debug_local.py

@@ -6,7 +6,14 @@
 
 root_directory = Path(__file__).resolve().parent.parent
 sys.path.insert(0, str(root_directory / "src"))
-from fabric_cicd import FabricWorkspace, publish_all_items, unpublish_all_orphan_items
+from fabric_cicd import (
+    FabricWorkspace,
+    publish_all_items,
+    unpublish_all_orphan_items,
+    enable_debug_log,
+)
+
+# enable_debug_log()
 
 # The defined environment values should match the names found in the parameter.yml file
 workspace_id = "8f5c0cec-a8ea-48cd-9da4-871dc2642f4c"
@@ -34,8 +41,6 @@
     item_type_in_scope=item_type_in_scope,
     # Override base url in rare cases where it's different
     base_api_url="https://msitapi.fabric.microsoft.com/",
-    # Print all api calls to debug what is being passed to fabric
-    # debug_output=True,
     token_credential=token_credential,
 )
 

pyproject.toml

@@ -21,11 +21,12 @@ license = { text = "MIT License" }
 dependencies = [
     "requests>=2.32.3",
     "pyyaml>=6.0.2",
-    "azure-identity>=1.19.0"
+    "azure-identity>=1.19.0",
+    "colorlog>=6.9.0"
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
 
 [project.urls]
-Repository = "https://github.com/microsoft/fabric-cicd.git"
+Repository = "https://github.com/microsoft/fabric-cicd.git"

sample/optional_parameters.py

@@ -2,7 +2,12 @@
 Example of optional parameters for FabricWorkspace and publish functions.
 """
 
-from fabric_cicd import FabricWorkspace, publish_all_items, unpublish_all_orphan_items
+from fabric_cicd import (
+    FabricWorkspace,
+    publish_all_items,
+    unpublish_all_orphan_items,
+    enable_debug_log,
+)
 
 # Sample values for FabricWorkspace parameters
 workspace_id = "your-workspace-id"
@@ -12,6 +17,9 @@
 base_api_url = "https://msitapi.fabric.microsoft.com/"
 token_credential = TokenCredential
 
+# Optional: Print all API calls to log file
+enable_debug_log()
+
 # Initialize the FabricWorkspace object with the required and optional parameters
 target_workspace = FabricWorkspace(
     workspace_id=workspace_id,
@@ -22,8 +30,6 @@
     base_api_url=base_api_url,
     # Optional: Override token credential to use a different authentication
     token_credential=None,
-    # Optional: Print all API calls to debug what is being passed to Fabric
-    debug_output=True,
 )
 
 # Publish all items defined in item_type_in_scope

src/fabric_cicd/__init__.py

@@ -1,5 +1,74 @@
-from fabric_cicd.fabric_workspace import FabricWorkspace
+"""
+Initializes the fabric_cicd package and sets up logging.
+"""
 
+from fabric_cicd.fabric_workspace import FabricWorkspace
 from fabric_cicd.publish import publish_all_items, unpublish_all_orphan_items
+import logging
+import colorlog
+
+
+def _configure_logger(level: int = logging.INFO) -> None:
+    """
+    Configure the logger.
+
+    :param level: The log level to set. Must be one of the standard logging levels.
+    """
+    # Configure logging
+    log_formatter = "%(asctime)s - %(levelname)s - %(name)s - %(message)s"
+
+    logging.basicConfig(
+        level=(
+            # For non-fabric_cicd packages: INFO if DEBUG, else ERROR
+            logging.INFO
+            if level == logging.DEBUG
+            else logging.ERROR
+        ),
+        format=log_formatter,
+        filename="fabric_cicd.error.log",
+        filemode="w",
+    )
+
+    package_logger = logging.getLogger("fabric_cicd")
+    package_logger.setLevel(level)
+    package_logger.handlers = []
+
+    # Configure Console Handler
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(level)
+
+    console_formatter = colorlog.ColoredFormatter(
+        "%(log_color)s[%(levelname)s] %(asctime)s - %(message)s",
+        datefmt="%H:%M:%S",
+        log_colors={
+            "DEBUG": "cyan",
+            "INFO": "green",
+            "WARNING": "yellow",
+            "ERROR": "red",
+            "CRITICAL": "bold_red",
+        },
+    )
+    console_handler.setFormatter(console_formatter)
+
+    # Add the handler to the logger
+    package_logger.addHandler(console_handler)
+
+
+# Initial logger configuration
+_configure_logger()
+
+
+def enable_debug_log() -> None:
+    """
+    Set the log level for all loggers within the fabric_cicd package to DEBUG.
+    """
+
+    _configure_logger(logging.DEBUG)
+
 
-__all__ = ["FabricWorkspace", "publish_all_items", "unpublish_all_orphan_items"]
+__all__ = [
+    "FabricWorkspace",
+    "publish_all_items",
+    "unpublish_all_orphan_items",
+    "enable_debug_log",
+]

src/fabric_cicd/_common/_logging.py → src/fabric_cicd/_common/_custom_print.py

@@ -1,34 +1,4 @@
 import datetime
-import logging
-import os
-import sys
-
-_log_file_name = "fabric_cicd.error.log"
-_log_file_path = os.path.join(os.path.dirname(__file__), _log_file_name)
-
-# Configure logging
-logging.basicConfig(
-    level=logging.ERROR,
-    format="%(asctime)s - %(levelname)s - %(message)s",
-    filename=_log_file_name,  # Specify the log file name
-    filemode="w",
-)
-
-
-def log_exceptions(exc_type, exc_value, exc_traceback):
-    if issubclass(exc_type, KeyboardInterrupt):
-        sys.__excepthook__(exc_type, exc_value, exc_traceback)
-        return
-    message = f"{exc_type.__name__}: {exc_value}"
-    logging.error(message, exc_info=(exc_type, exc_value, exc_traceback))
-
-    # Prevent terminal output
-    sys.stderr = open(os.devnull, "w")
-    print_line(f"{message}\n\nRefer to {_log_file_path} for full log", "red")
-
-
-# Set the custom exception hook
-sys.excepthook = log_exceptions
 
 
 def _get_ansi_color(color_name):
@@ -108,47 +78,3 @@ def print_header(message, color="green"):
     print_line(formatted_message, color)
     print_line(line_separator, color)
     print_line("")  # Print a blank line after the header
-
-
-def print_info_line(message):
-    """
-    Prints an error message
-
-    :param message: The message to print.
-    """
-    color = "yellow"
-
-    print_line(f"Info: {message}", color)
-
-
-class InputError(Exception):
-    def __init__(self, message):
-        super().__init__(message)
-
-
-class TokenError(Exception):
-    def __init__(self, message):
-        super().__init__(message)
-
-
-class ParsingError(Exception):
-    def __init__(self, message):
-        super().__init__(message)
-
-
-class InvokeError(Exception):
-    def __init__(self, message, response, method, url, body, debug_output=False):
-        super().__init__(message)
-        if debug_output:
-            debug_color = "gray"
-            print_header("DEBUG OUTPUT", debug_color)
-            print_line(f"URL: {url}", debug_color)
-            print_line(f"Method: {method}", debug_color)
-            print_line(f"Request Body: {body}", debug_color)
-            if response is not None:
-                print_line(f"Response Status: {response.status_code}", debug_color)
-                print_line("Response Header:", debug_color)
-                print_line(response.headers, debug_color)
-                print_line("Response Body:", debug_color)
-                print_line(response.text, debug_color)
-                print_line("")

src/fabric_cicd/_common/_exceptions.py

@@ -0,0 +1,59 @@
+import json
+import logging
+
+
+class ParsingError(Exception):
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class InputError(Exception):
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class TokenError(Exception):
+    def __init__(self, message):
+        super().__init__(message)
+
+
+class InvokeError(Exception):
+    def __init__(self, message, response, method, url, body, logger):
+        super().__init__(message)
+        log_invoke_payload(logger, response, method, url, body, error=True)
+
+
+def log_invoke_payload(logger, response, method, url, body, error=False):
+    debug_message = [
+        f"\nURL: {url}",
+        f"Method: {method}",
+        (
+            f"Request Body:\n{json.dumps(body, indent=4)}"
+            if body
+            else "Request Body: None"
+        ),
+    ]
+    if response is not None:
+        debug_message.extend(
+            [
+                f"Response Status: {response.status_code}",
+                "Response Headers:",
+                json.dumps(dict(response.headers), indent=4),
+                "Response Body:",
+                (
+                    json.dumps(response.json(), indent=4)
+                    if response.headers.get("Content-Type") == "application/json"
+                    else response.text
+                ),
+                "",
+            ]
+        )
+
+    # Join all parts into a single message
+    full_debug_message = "\n".join(debug_message)
+    if error:
+        logger.error(full_debug_message)
+    elif logger.isEnabledFor(logging.DEBUG):
+        logger.debug(full_debug_message)
+    else:
+        logger.info(full_debug_message)