Source code for swh.deposit.loader.checker

# Copyright (C) 2017-2020  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information

import logging
from typing import Any, Dict

import sentry_sdk

from swh.core import config
from swh.deposit.client import PrivateApiDepositClient

logger = logging.getLogger(__name__)

[docs] class DepositChecker: """Deposit checker implementation. Trigger deposit's checks through the private api. """ def __init__(self): self.config: Dict[str, Any] = config.load_from_envvar() self.client = PrivateApiDepositClient(config=self.config["deposit"])
[docs] def check(self, collection: str, deposit_id: str) -> Dict[str, str]: status = None deposit_check_url = f"/{collection}/{deposit_id}/check/" logger.debug("deposit-check-url: %s", deposit_check_url) try: r = self.client.check(deposit_check_url) logger.debug("Check result: %s", r) status = "eventful" if r == "verified" else "failed" except Exception: logger.exception("Failure during check on '%s'", deposit_check_url) sentry_sdk.capture_exception() status = "failed" logger.debug("Check status: %s", status) return {"status": status}