def purge_executions(logger, timestamp, action_ref=None, purge_incomplete=False):
    """
    Purge action executions and corresponding live action, execution output objects.

    :param timestamp: Exections older than this timestamp will be deleted.
    :type timestamp: ``datetime.datetime

    :param action_ref: Only delete executions for the provided actions.
    :type action_ref: ``str``

    :param purge_incomplete: True to also delete executions which are not in a done state.
    :type purge_incomplete: ``bool``
    """
    if not timestamp:
        raise ValueError("Specify a valid timestamp to purge.")

    logger.info(
        "Purging executions older than timestamp: %s"
        % timestamp.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    )

    filters = {}

    if purge_incomplete:
        filters["start_timestamp__lt"] = timestamp
    else:
        filters["end_timestamp__lt"] = timestamp
        filters["start_timestamp__lt"] = timestamp
        filters["status"] = {"$in": DONE_STATES}

    exec_filters = copy.copy(filters)
    if action_ref:
        exec_filters["action__ref"] = action_ref

    liveaction_filters = copy.deepcopy(filters)
    if action_ref:
        liveaction_filters["action"] = action_ref

    to_delete_execution_dbs = []

    # 1. Delete ActionExecutionDB objects
    try:
        # Note: We call list() on the query set object because it's lazyily evaluated otherwise
        to_delete_execution_dbs = list(
            ActionExecution.query(
                only_fields=["id"], no_dereference=True, **exec_filters
            )
        )
        deleted_count = ActionExecution.delete_by_query(**exec_filters)
    except InvalidQueryError as e:
        msg = (
            "Bad query (%s) used to delete execution instances: %s"
            "Please contact support."
            % (
                exec_filters,
                six.text_type(e),
            )
        )
        raise InvalidQueryError(msg)
    except:
        logger.exception(
            "Deletion of execution models failed for query with filters: %s.",
            exec_filters,
        )
    else:
        logger.info("Deleted %s action execution objects" % (deleted_count))

    # 2. Delete LiveActionDB objects
    try:
        deleted_count = LiveAction.delete_by_query(**liveaction_filters)
    except InvalidQueryError as e:
        msg = (
            "Bad query (%s) used to delete liveaction instances: %s"
            "Please contact support."
            % (
                liveaction_filters,
                six.text_type(e),
            )
        )
        raise InvalidQueryError(msg)
    except:
        logger.exception(
            "Deletion of liveaction models failed for query with filters: %s.",
            liveaction_filters,
        )
    else:
        logger.info("Deleted %s liveaction objects" % (deleted_count))

    # 3. Delete ActionExecutionOutputDB objects
    to_delete_exection_ids = [
        str(execution_db.id) for execution_db in to_delete_execution_dbs
    ]
    output_dbs_filters = {}
    output_dbs_filters["execution_id"] = {"$in": to_delete_exection_ids}

    try:
        deleted_count = ActionExecutionOutput.delete_by_query(**output_dbs_filters)
    except InvalidQueryError as e:
        msg = (
            "Bad query (%s) used to delete execution output instances: %s"
            "Please contact support." % (output_dbs_filters, six.text_type(e))
        )
        raise InvalidQueryError(msg)
    except:
        logger.exception(
            "Deletion of execution output models failed for query with filters: %s.",
            output_dbs_filters,
        )
    else:
        logger.info("Deleted %s execution output objects" % (deleted_count))

    zombie_execution_instances = len(
        ActionExecution.query(only_fields=["id"], no_dereference=True, **exec_filters)
    )
    zombie_liveaction_instances = len(
        LiveAction.query(only_fields=["id"], no_dereference=True, **liveaction_filters)
    )

    if (zombie_execution_instances > 0) or (zombie_liveaction_instances > 0):
        logger.error("Zombie execution instances left: %d.", zombie_execution_instances)
        logger.error(
            "Zombie liveaction instances left: %s.", zombie_liveaction_instances
        )

    # Print stats
    logger.info("All execution models older than timestamp %s were deleted.", timestamp)

    Purge action executions and corresponding live action, execution output objects.

    :param timestamp: Exections older than this timestamp will be deleted.
    :type timestamp: ``datetime.datetime

    :param action_ref: Only delete executions for the provided actions.
    :type action_ref: ``str``

    :param purge_incomplete: True to also delete executions which are not in a done state.
    :type purge_incomplete: ``bool``
    
st2common/st2common/garbage_collection/executions.py