Class BundleCredentials

plugin/app/models/credentials.py:9–70

class BundleCredentials(BaseModel):

    credentials: Dict[str, str] = Field({})

    def __getattr__(self, item):
        # This method is called only if Python doesn't find the attribute
        # in the usual places (instance dictionary or class hierarchy).
        if item in self.credentials:
            return self.credentials.get(item)
        else:
            # If the item is not in credentials return None
            return None

    def encrypt(self):
        for key, value in self.credentials.items():
            if value is not None:
                encrypted_value = aes_encrypt(value)
                self.credentials[key] = str(encrypted_value)
        return self

    def decrypt(self):
        for key, value in self.credentials.items():
            if value is not None:
                if "," in value:
                    try:
                        decrypted_value = aes_decrypt(value)
                        self.credentials[key] = str(decrypted_value)
                    except ValueError as e:
                        raise ValueError(f"error decrypting {key}: {e}")
                else:
                    raise ValueError(f"invalid credentials. Not encrypted: {value}")
        return self

    def load_input(self, bundle_id, credentials: Dict[str, str]):
        try:
            from app.cache import get_bundle

            allowed_fields = get_bundle(bundle_id).allowed_credential_names()
            for field in allowed_fields:
                if field in credentials:
                    self.credentials[field] = credentials[field]
            return self
        except Exception as e:
            raise ValueError(f"error loading credentials from input: {e}")

    def load_default(self, bundle_id):
        try:
            from app.cache import get_bundle

            default_credentials = get_bundle(bundle_id).allowed_credential_names()
            for name in default_credentials:
                self.credentials[name] = load_str_env(name, required=True)
            return self
        except Exception as e:
            raise ValueError(f"error loading credentials from env: {e}")

    def to_dict(self, bundle_id):
        from app.cache import get_bundle