(data: Dict)
| 71 | |
| 72 | |
| 73 | def validate_bundle_credentials(data: Dict) -> BundleCredentials: |
| 74 | from app.cache import get_bundle |
| 75 | |
| 76 | bundle_id = data.get("bundle_id") |
| 77 | if bundle_id and isinstance(bundle_id, str): |
| 78 | if not get_bundle(bundle_id): |
| 79 | raise ValueError(f"provider {bundle_id} does not exist.") |
| 80 | else: |
| 81 | raise ValueError("bundle_id is required.") |
| 82 | |
| 83 | # handle credentials |
| 84 | if data.get("credentials") and data.get("encrypted_credentials"): |
| 85 | raise ValueError("either credentials or encrypted_credentials is required, but not both.") |
| 86 | |
| 87 | credentials = BundleCredentials() |
| 88 | |
| 89 | if data.get("credentials"): |
| 90 | if not isinstance(data.get("credentials"), dict): |
| 91 | raise ValueError("credentials must be a dict.") |
| 92 | credentials.load_input(bundle_id, data.get("credentials")) |
| 93 | elif data.get("encrypted_credentials"): |
| 94 | if not isinstance(data.get("encrypted_credentials"), dict): |
| 95 | raise ValueError("encrypted_credentials must be a dict.") |
| 96 | credentials.load_input(bundle_id, data.get("encrypted_credentials")) |
| 97 | credentials.decrypt() |
| 98 | else: |
| 99 | try: |
| 100 | credentials.load_default(bundle_id) |
| 101 | except Exception as e: |
| 102 | raise ValueError(f"Failed to load default credentials for provider {bundle_id} from env") |
| 103 | |
| 104 | return credentials |
no test coverage detected