processors

processors

Custom logging processors for crew_dcs.

This module contains result processors and extractors specifically designed for crew_dcs components to provide better logging integration.

DomoEntityExtractor

Bases: EntityExtractor

Custom entity extractor for Domo routes that extracts entity info from function parameters.

extract

extract(
    func: Any, args: tuple, kwargs: dict
) -> LogEntity | None

Extract entity information from function parameters.

This extractor looks at the function parameters to determine what type of entity is being accessed and extracts relevant information.

Source code in src/crew_dcs/utils/logging/processors.py

def extract(self, func: Any, args: tuple, kwargs: dict) -> LogEntity | None:
    """Extract entity information from function parameters.

    This extractor looks at the function parameters to determine what type of entity
    is being accessed and extracts relevant information.
    """
    # Extract entity type from function name
    func_name = func.__name__.lower()

    # Determine entity type based on function name patterns
    if "dataset" in func_name or "datasource" in func_name:
        return self._extract_dataset_entity(kwargs)
    if "card" in func_name:
        return self._extract_card_entity(kwargs)
    if "user" in func_name:
        return self._extract_user_entity(kwargs)
    if "page" in func_name or "stack" in func_name:
        return self._extract_page_entity(kwargs)
    if "auth" in func_name:
        return self._extract_auth_entity(kwargs)

    return None

DomoEntityObjectProcessor

Bases: ResultProcessor

Custom result processor for DomoEntity objects returned from class methods.

process

process(
    result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]

Process DomoEntity result and extract entity information.

Parameters:

Name	Type	Description	Default
result	Any	The function result (should be DomoEntity_w_Lineage)	required
http_details	HTTPDetails | None	Optional HTTP details to update	None

Returns:

Type	Description
tuple[dict[str, Any], HTTPDetails | None]	Tuple of (result_context dict with entity info, updated http_details)

Source code in src/crew_dcs/utils/logging/processors.py

def process(
    self, result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]:
    """Process DomoEntity result and extract entity information.

    Args:
        result: The function result (should be DomoEntity_w_Lineage)
        http_details: Optional HTTP details to update

    Returns:
        Tuple of (result_context dict with entity info, updated http_details)
    """
    result_context = {}

    # Extract entity information from DomoEntity object
    entity = self._extract_entity_from_domo_object(result)
    if entity:
        # Override the entity field with our extracted entity information
        # This will replace the decorator's entity field
        result_context["entity"] = {
            "type": entity.type,
            "id": entity.id,
            "name": entity.name,
            "additional_info": entity.additional_info,
        }

    return result_context, http_details

DomoEntityProcessor

Bases: ResultProcessor

Custom result processor for DomoEntity objects from route responses.

process

process(
    result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]

Process route result and extract entity information.

Parameters:

Name	Type	Description	Default
result	Any	The function result (should be ResponseGetData)	required
http_details	HTTPDetails | None	Optional HTTP details to update	None

Returns:

Type	Description
tuple[dict[str, Any], HTTPDetails | None]	Tuple of (result_context dict with entity info, updated http_details)

Source code in src/crew_dcs/utils/logging/processors.py

def process(  # noqa: C901
    self, result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]:
    """Process route result and extract entity information.

    Args:
        result: The function result (should be ResponseGetData)
        http_details: Optional HTTP details to update

    Returns:
        Tuple of (result_context dict with entity info, updated http_details)
    """
    result_context = {}

    # Debug: Print what we're processing
    print(f"DEBUG DomoEntityProcessor: Processing result type: {type(result)}")
    if hasattr(result, "request_metadata") and result.request_metadata:
        print(f"DEBUG DomoEntityProcessor: URL: {result.request_metadata.url}")

    # Extract entity information
    entity = self._extract_entity_info(result)
    if entity:
        print(f"DEBUG DomoEntityProcessor: Extracted entity: {entity}")
        # Override the entity field directly - this should work since result_context is spread after log_context
        result_context["entity"] = entity
    else:
        print("DEBUG DomoEntityProcessor: No entity extracted")

    # Update HTTP details if it's a ResponseGetData object
    if isinstance(result, rgd.ResponseGetData) and http_details:
        http_details.status_code = result.status

        # Extract response size and body
        if hasattr(result, "response"):
            response = result.response
            if isinstance(response, str | bytes):
                http_details.response_size = len(response)
                response_str = str(response)
                http_details.response_body = (
                    response_str[:500] if len(response_str) > 500 else response_str
                )
            elif isinstance(response, dict):
                # For dictionaries, show key information
                http_details.response_size = len(str(response))
                # Show first few keys and values for context
                keys = list(response.keys())[:5]
                summary = {k: response[k] for k in keys if k in response}
                http_details.response_body = summary
            elif hasattr(response, "__len__"):
                try:
                    response_len = len(response)
                except TypeError:
                    response_len = None
                else:
                    http_details.response_size = response_len
                    http_details.response_body = (
                        f"<{type(response).__name__} with {response_len} items>"
                    )
            else:
                http_details.response_body = f"<{type(response).__name__}>"

        # Use request metadata if available
        if hasattr(result, "request_metadata") and result.request_metadata:
            metadata = result.request_metadata
            if not http_details.url:
                http_details.url = metadata.url
            if not http_details.method:
                http_details.method = metadata.method
            if not http_details.headers:
                http_details.headers = metadata.headers
            if not http_details.params:
                http_details.params = metadata.params
            if not http_details.request_body:
                http_details.request_body = metadata.body

    return result_context, http_details

DomoEntityResultProcessor

Bases: ResultProcessor

Enhanced result processor that extracts rich entity information from DomoEntity objects.

process

process(
    result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]

Process the result to extract rich entity information from DomoEntity objects.

Source code in src/crew_dcs/utils/logging/processors.py

def process(
    self, result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]:
    """Process the result to extract rich entity information from DomoEntity objects."""
    result_context = {}

    # Try to extract entity information from the result
    entity_info = self._extract_rich_entity_info(result)
    if entity_info:
        # Put rich entity information in a custom field to complement basic entity
        result_context["domo_entity_info"] = entity_info

    # Update HTTP details if it's a ResponseGetData object
    if isinstance(result, rgd.ResponseGetData) and http_details:
        http_details.status_code = result.status

    return result_context, http_details

NoOpEntityExtractor

Bases: EntityExtractor

No-op entity extractor that returns None to avoid conflicts.

extract

extract(
    func: Any, args: tuple, kwargs: dict
) -> LogEntity | None

Return None to avoid entity conflicts.

Source code in src/crew_dcs/utils/logging/processors.py

def extract(self, func: Any, args: tuple, kwargs: dict) -> LogEntity | None:
    """Return None to avoid entity conflicts."""
    return None

ResponseGetDataProcessor

Bases: ResultProcessor

Custom result processor for ResponseGetData objects.

process

process(
    result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]

Process ResponseGetData result and update HTTP details.

Parameters:

Name	Type	Description	Default
result	Any	The function result (should be ResponseGetData)	required
http_details	HTTPDetails | None	Optional HTTP details to update	None

Returns:

Type	Description
tuple[dict[str, Any], HTTPDetails | None]	Tuple of (result_context dict, updated http_details)

Source code in src/crew_dcs/utils/logging/processors.py

def process(
    self, result: Any, http_details: HTTPDetails | None = None
) -> tuple[dict[str, Any], HTTPDetails | None]:
    """Process ResponseGetData result and update HTTP details.

    Args:
        result: The function result (should be ResponseGetData)
        http_details: Optional HTTP details to update

    Returns:
        Tuple of (result_context dict, updated http_details)
    """
    result_context = {}

    if isinstance(result, rgd.ResponseGetData) and http_details:
        # Update HTTP details with response information
        http_details.status_code = result.status

        # Extract response size and body
        if hasattr(result, "response"):
            response = result.response
            http_details.response_body = self._format_response_body(response)

            # Calculate response size
            if isinstance(response, str | bytes):
                http_details.response_size = len(response)
            elif hasattr(response, "__len__"):
                try:
                    http_details.response_size = len(response)
                except TypeError:
                    http_details.response_size = None

        # Use request metadata if available to fill in missing request details
        if hasattr(result, "request_metadata") and result.request_metadata:
            metadata = result.request_metadata
            if not http_details.url:
                http_details.url = metadata.url
            if not http_details.method:
                http_details.method = metadata.method
            if not http_details.headers:
                # Sanitize headers before setting
                http_details.headers = self._sanitize_headers(metadata.headers)
            if not http_details.params:
                http_details.params = metadata.params
            if not http_details.request_body:
                http_details.request_body = metadata.body

    return result_context, http_details