def _update_call_state()

in lca-ai-stack/source/lambda_functions/call_event_stream_processor/tumbling_window_state/call_state_manager.py [0:0]

• 32 lines of code
• 6 McCabe index (conditional complexity)

    def _update_call_state(previous: StatePerCallId, current: Dict[str, Any]) -> StatePerCallId:
        """Updates call status and created/updated dates"""
        LOGGER.debug(
            "update call status - previous current",
            extra=dict(previous=previous, current=current),
        )
        now = datetime.now(timezone.utc).astimezone().isoformat()

        created_at = previous.get("CreatedAt", now)

        # take the latest UpdatedAt
        current_updated_at = current.get("UpdatedAt", now)
        if not isinstance(current_updated_at, str):
            current_updated_at = now
        previous_updated_at = previous.get("UpdatedAt", "")
        if not isinstance(previous_updated_at, str):
            previous_updated_at = ""
        updated_at = (
            current_updated_at if current_updated_at >= previous_updated_at else previous_updated_at
        )

        call_state = current if current_updated_at >= previous_updated_at else previous
        current_status = current.get("Status", "STARTED")
        previous_status = previous.get("Status", "STARTED")
        # do not override an ENDED status
        status = (
            "ENDED"
            if "ENDED" in {current_status, previous_status}
            else call_state.get("Status", "STARTED")
        )

        updated_call_state: StatePerCallId = {
            **previous,  # type: ignore
            "CreatedAt": created_at,
            "UpdatedAt": updated_at,
            "Status": status,
        }

        LOGGER.debug("updated call state", extra=dict(updated_call_state=updated_call_state))

        return updated_call_state