Module rugged.workers.snapshot-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 = 'snapshot-worker'
log = get_rugged_logger()


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

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

Classes

class SnapshotWorker

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

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

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

Ancestors

Methods

def update_snapshot_task(**context)

Task to update snapshot metadata for a TUF repository.

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

Inherited members