Module rugged.workers.timestamp-worker

Expand source code 
from rugged.lib.logger import (
    get_rugged_logger,
    set_log_level_from_context,
)
from rugged.lib.task_queue import TaskQueue
from rugged.tuf.repo import RuggedRepository
from rugged.workers.base_worker import BaseWorker

worker = TaskQueue().get_task_queue()
queue = 'timestamp-worker'
log = get_rugged_logger()


class TimeStampWorker(BaseWorker):
    """ Rugged (Celery) worker that fulfills the TUF 'timestamp' role. """

    @staticmethod
    @worker.task(name='update_timestamp_task', queue=queue)
    def update_timestamp_task(**context):
        """ Task to update timestamp metadata for a TUF repository. """
        set_log_level_from_context(context)
        log.info("Received update-timestamp task.")
        repo = RuggedRepository()
        repo.load()
        repo.update_timestamp()
        result = repo.write_metadata('timestamp')
        if result:
            message = "Updated timestamp metadata."
            log.info(message)
        else:
            message = "Failed to timestamp snapshot metadata."
            log.error(message)
        return (result, message)

Classes

class TimeStampWorker

Rugged (Celery) worker that fulfills the TUF 'timestamp' role.

Expand source code 
class TimeStampWorker(BaseWorker):
    """ Rugged (Celery) worker that fulfills the TUF 'timestamp' role. """

    @staticmethod
    @worker.task(name='update_timestamp_task', queue=queue)
    def update_timestamp_task(**context):
        """ Task to update timestamp metadata for a TUF repository. """
        set_log_level_from_context(context)
        log.info("Received update-timestamp task.")
        repo = RuggedRepository()
        repo.load()
        repo.update_timestamp()
        result = repo.write_metadata('timestamp')
        if result:
            message = "Updated timestamp metadata."
            log.info(message)
        else:
            message = "Failed to timestamp snapshot metadata."
            log.error(message)
        return (result, message)

Ancestors

Static methods

def update_timestamp_task(**context)

Task to update timestamp metadata for a TUF repository.

Expand source code 
@staticmethod
@worker.task(name='update_timestamp_task', queue=queue)
def update_timestamp_task(**context):
    """ Task to update timestamp metadata for a TUF repository. """
    set_log_level_from_context(context)
    log.info("Received update-timestamp task.")
    repo = RuggedRepository()
    repo.load()
    repo.update_timestamp()
    result = repo.write_metadata('timestamp')
    if result:
        message = "Updated timestamp metadata."
        log.info(message)
    else:
        message = "Failed to timestamp snapshot metadata."
        log.error(message)
    return (result, message)

Inherited members