annotate env/lib/python3.9/site-packages/cwltool/secrets.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

"""Minimal in memory storage of secrets."""
import uuid
from typing import Dict, List, MutableMapping, MutableSequence, Optional, cast

from .utils import CWLObjectType, CWLOutputType


class SecretStore:
    """Minimal implementation of a secret storage."""

    def __init__(self) -> None:
        """Initialize the secret store."""
        self.secrets = {}  # type: Dict[str, str]

    def add(self, value: Optional[CWLOutputType]) -> Optional[CWLOutputType]:
        """
        Add the given value to the store.

        Returns a placeholder value to use until the real value is needed.
        """
        if not isinstance(value, str):
            raise Exception("Secret store only accepts strings")

        if value not in self.secrets:
            placeholder = "(secret-%s)" % str(uuid.uuid4())
            self.secrets[placeholder] = value
            return placeholder
        return value

    def store(self, secrets: List[str], job: CWLObjectType) -> None:
        """Sanitize the job object of any of the given secrets."""
        for j in job:
            if j in secrets:
                job[j] = self.add(job[j])

    def has_secret(self, value: CWLOutputType) -> bool:
        """Test if the provided document has any of our secrets."""
        if isinstance(value, str):
            for k in self.secrets:
                if k in value:
                    return True
        elif isinstance(value, MutableMapping):
            for this_value in value.values():
                if self.has_secret(cast(CWLOutputType, this_value)):
                    return True
        elif isinstance(value, MutableSequence):
            for this_value in value:
                if self.has_secret(cast(CWLOutputType, this_value)):
                    return True
        return False

    def retrieve(self, value: CWLOutputType) -> CWLOutputType:
        """Replace placeholders with their corresponding secrets."""
        if isinstance(value, str):
            for key, this_value in self.secrets.items():
                value = value.replace(key, this_value)
            return value
        elif isinstance(value, MutableMapping):
            return {k: self.retrieve(cast(CWLOutputType, v)) for k, v in value.items()}
        elif isinstance(value, MutableSequence):
            return [self.retrieve(cast(CWLOutputType, v)) for v in value]
        return value