perf: Parallelize all items by central encapsulation into base classes

.gitignore

@@ -164,4 +164,5 @@ cython_debug/
 
 # http traces should only be committed at the fixture root
 /http_trace.json
+/http_trace.json.lock
 /http_trace.json.gz

src/fabric_cicd/__init__.py

@@ -9,6 +9,7 @@
 import fabric_cicd.constants as constants
 from fabric_cicd._common._check_utils import check_version
 from fabric_cicd._common._logging import configure_logger, exception_handler
+from fabric_cicd.constants import FeatureFlag, ItemType
 from fabric_cicd.fabric_workspace import FabricWorkspace
 from fabric_cicd.publish import deploy_with_config, publish_all_items, unpublish_all_orphan_items
 
@@ -56,6 +57,8 @@ def change_log_level(level: str = "DEBUG") -> None:
 
 __all__ = [
     "FabricWorkspace",
+    "FeatureFlag",
+    "ItemType",
     "append_feature_flag",
     "change_log_level",
     "deploy_with_config",

src/fabric_cicd/_common/_file_lock.py

@@ -0,0 +1,64 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Cross-platform file locking."""
+
+import sys
+from pathlib import Path
+from types import TracebackType
+from typing import Callable, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+class FileLock:
+    """File lock context manager."""
+
+    def __init__(self, lock_file: str) -> None:
+        self.lock_path = Path(f"{lock_file}.lock")
+        self._lock_file: Optional[object] = None
+
+    def __enter__(self) -> "FileLock":
+        self._lock_file = self.lock_path.open("w")
+        if sys.platform == "win32":
+            import msvcrt
+
+            msvcrt.locking(self._lock_file.fileno(), msvcrt.LK_LOCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_EX)
+        return self
+
+    def __exit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> bool:
+        if self._lock_file:
+            if sys.platform == "win32":
+                import msvcrt
+
+                msvcrt.locking(self._lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+            else:
+                import fcntl
+
+                fcntl.flock(self._lock_file.fileno(), fcntl.LOCK_UN)
+            self._lock_file.close()
+        return False
+
+    @staticmethod
+    def run_with_lock(lock_file: str, func: Callable[[], T]) -> T:
+        """
+        Execute a function while holding an exclusive file lock.
+
+        Args:
+            lock_file: Path to the file to lock (a .lock suffix will be added)
+            func: The function to execute while holding the lock
+
+        Returns:
+            The return value of the function
+        """
+        with FileLock(lock_file):
+            return func()

src/fabric_cicd/_common/_http_tracer.py

@@ -16,6 +16,7 @@
 
 import requests
 
+from fabric_cicd._common._file_lock import FileLock
 from fabric_cicd.constants import AUTHORIZATION_HEADER, EnvVar
 
 logger = logging.getLogger(__name__)
@@ -192,40 +193,43 @@ def save(self) -> None:
             return
 
         try:
-            output_path = Path(self.output_file)
-            existing_traces: list[dict] = []
-            if output_path.exists():
-                with output_path.open("r") as f:
-                    existing_data = json.load(f)
-                    existing_traces = existing_data.get("traces", [])
-
-            for capture in self.captures:
-                request_b64 = capture.get("request_b64", "")
-                response_b64 = capture.get("response_b64", "")
-
-                request_data = None
-                response_data = None
-
-                if request_b64:
-                    request_data = json.loads(base64.b64decode(request_b64).decode())
-                if response_b64:
-                    response_data = json.loads(base64.b64decode(response_b64).decode())
-
-                existing_traces.append({"request": request_data, "response": response_data})
-
-            existing_traces.sort(key=lambda x: x["request"].get("timestamp", "") if x.get("request") else "")
-            output_data = {
-                "description": "HTTP trace data from Fabric API interactions",
-                "total_traces": len(existing_traces),
-                "traces": existing_traces,
-            }
-
-            with output_path.open("w") as f:
-                json.dump(output_data, f, indent=2)
-
+            FileLock.run_with_lock(self.output_file, self._flush_traces_to_file)
         except Exception as e:
             logger.warning(f"Failed to save HTTP trace: {e}")
 
+    def _flush_traces_to_file(self) -> None:
+        """Flush captured traces to the output file (called within lock)."""
+        output_path = Path(self.output_file)
+        existing_traces: list[dict] = []
+        if output_path.exists() and output_path.stat().st_size > 0:
+            with output_path.open("r") as f:
+                existing_data = json.load(f)
+                existing_traces = existing_data.get("traces", [])
+
+        for capture in self.captures:
+            request_b64 = capture.get("request_b64", "")
+            response_b64 = capture.get("response_b64", "")
+
+            request_data = None
+            response_data = None
+
+            if request_b64:
+                request_data = json.loads(base64.b64decode(request_b64).decode())
+            if response_b64:
+                response_data = json.loads(base64.b64decode(response_b64).decode())
+
+            existing_traces.append({"request": request_data, "response": response_data})
+
+        existing_traces.sort(key=lambda x: x["request"].get("timestamp", "") if x.get("request") else "")
+        output_data = {
+            "description": "HTTP trace data from Fabric API interactions",
+            "total_traces": len(existing_traces),
+            "traces": existing_traces,
+        }
+
+        with output_path.open("w") as f:
+            json.dump(output_data, f, indent=2)
+
 
 class HTTPTracerFactory:
     """Factory class for creating HTTP tracer instances."""

src/fabric_cicd/_common/_logging.py

@@ -10,7 +10,7 @@
 import traceback
 from logging import LogRecord
 from pathlib import Path
-from typing import ClassVar
+from typing import Any, ClassVar, Optional
 
 from fabric_cicd import constants
 from fabric_cicd._common import _exceptions
@@ -38,12 +38,20 @@ def format(self, record: LogRecord) -> str:
 
         level_name = f"{level_color}[{level_name}]"
         timestamp = f"{self.formatTime(record, self.datefmt)}"
+        item_context = ""
+        item_type = getattr(record, "item_type", None)
+        item_name = getattr(record, "item_name", None)
+        if item_type or item_name:
+            type_str = item_type if item_type else "?"
+            name_str = item_name if item_name else "?"
+            item_context = f"[{type_str}:{name_str}] "
+
         message = f"{record.getMessage()}{Style.RESET_ALL}"
 
         # indent if the message contains "->"
         if constants.INDENT in message:
             message = message.replace(constants.INDENT, "")
-            full_message = f"{' ' * 8} {timestamp} - {message}"
+            full_message = f"{' ' * 8} {timestamp} - {item_context}{message}"
         else:
             # Calculate visual length by removing ANSI escape codes
 
@@ -54,10 +62,109 @@ def format(self, record: LogRecord) -> str:
             # Pad to 16 visual characters
             padding = " " * max(0, 8 - visual_level_length)
 
-            full_message = f"{level_name}{padding} {timestamp} - {message}"
+            full_message = f"{level_name}{padding} {timestamp} - {item_context}{message}"
         return full_message
 
 
+class ItemLoggerAdapter(logging.LoggerAdapter):
+    """
+    A LoggerAdapter that automatically includes item context (type, name) in log messages.
+
+    This adapter is designed for use during item publishing operations where logs need
+    to be traceable to specific items, especially in parallel execution scenarios.
+
+    Example:
+        >>> logger = logging.getLogger(__name__)
+        >>> item_logger = ItemLoggerAdapter(logger, item_type="Notebook", item_name="MyNotebook")
+        >>> item_logger.info("Publishing")
+        # Output: [info] 14:32:01 - [_publish_item:650] - [Notebook>MyNotebook] Publishing
+    """
+
+    def __init__(
+        self,
+        logger: logging.Logger,
+        item_type: Optional[str] = None,
+        item_name: Optional[str] = None,
+    ) -> None:
+        """
+        Initialize the ItemLoggerAdapter with item context.
+
+        Args:
+            logger: The underlying logger to wrap.
+            item_type: The type of the item (e.g., "Notebook", "Environment").
+            item_name: The display name of the item.
+        """
+        extra = {
+            "item_type": item_type,
+            "item_name": item_name,
+        }
+        super().__init__(logger, extra)
+
+    def process(self, msg: str, kwargs: dict[str, Any]) -> tuple[str, dict[str, Any]]:
+        """
+        Process the logging call to inject item context into the extra dict.
+
+        Args:
+            msg: The log message.
+            kwargs: Keyword arguments passed to the logging call.
+
+        Returns:
+            Tuple of (message, kwargs) with item context injected.
+        """
+        extra = kwargs.get("extra", {})
+        extra.update(self.extra)
+        kwargs["extra"] = extra
+        return msg, kwargs
+
+    def with_item(
+        self,
+        item_type: Optional[str] = None,
+        item_name: Optional[str] = None,
+    ) -> "ItemLoggerAdapter":
+        """
+        Create a new ItemLoggerAdapter with updated item context.
+
+        This allows creating derived loggers with different item context
+        while reusing the same underlying logger.
+
+        Args:
+            item_type: The type of the item (overrides current if provided).
+            item_name: The display name of the item (overrides current if provided).
+
+        Returns:
+            A new ItemLoggerAdapter with the updated context.
+        """
+        return ItemLoggerAdapter(
+            self.logger,
+            item_type=item_type if item_type is not None else self.extra.get("item_type"),
+            item_name=item_name if item_name is not None else self.extra.get("item_name"),
+        )
+
+
+def get_item_logger(
+    name: str,
+    item_type: Optional[str] = None,
+    item_name: Optional[str] = None,
+) -> ItemLoggerAdapter:
+    """
+    Factory function to create an ItemLoggerAdapter for item-scoped logging.
+
+    Args:
+        name: The logger name (typically __name__).
+        item_type: The type of the item (e.g., "Notebook", "Environment").
+        item_name: The display name of the item.
+
+    Returns:
+        An ItemLoggerAdapter configured with the item context.
+
+    Example:
+        >>> item_logger = get_item_logger(__name__, item_type="Notebook", item_name="MyNotebook")
+        >>> item_logger.info("Starting publish")
+    """
+    logger = logging.getLogger(name)
+    return ItemLoggerAdapter(logger, item_type=item_type, item_name=item_name)
+
+
 def configure_logger(level: int = logging.INFO) -> None:
     """
     Configure the logger.