def purge_task_executions(logger, timestamp, purge_incomplete=False):
    """
    Purge task 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 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)
    try:
        deleted_count = TaskExecution.delete_by_query(**exec_filters)
    except InvalidQueryError as e:
        msg = (
            "Bad query (%s) used to delete task execution instances: %s"
            "Please contact support." % (exec_filters, six.text_type(e))
        )
        raise InvalidQueryError(msg)
    except:
        logger.exception(
            "Deletion of task execution models failed for query with filters: %s.",
            exec_filters,
        )
    else:
        logger.info("Deleted %s task execution objects" % deleted_count)

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

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

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

    Purge task 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