def purge_workflow_executions(logger, timestamp, purge_incomplete=False):
    """
    Purge workflow execution output objects.

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

    :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 workflow 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)

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

    zombie_execution_instances = len(
        WorkflowExecution.query(only_fields=["id"], no_dereference=True, **exec_filters)
    )

    if zombie_execution_instances > 0:
        logger.error(
            "Zombie workflow execution instances left: %d.", zombie_execution_instances
        )

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

    Purge workflow execution output objects.

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

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