(
model_schema_id: str,
provider_model_id: Optional[str],
model_type: Optional[str],
credentials: Optional[Dict],
encrypted_credentials: Optional[Dict],
properties: Optional[Dict],
)
| 10 | |
| 11 | |
| 12 | async def verify_credentials( |
| 13 | model_schema_id: str, |
| 14 | provider_model_id: Optional[str], |
| 15 | model_type: Optional[str], |
| 16 | credentials: Optional[Dict], |
| 17 | encrypted_credentials: Optional[Dict], |
| 18 | properties: Optional[Dict], |
| 19 | ) -> ResponseWrapper: |
| 20 | payload = { |
| 21 | "model_schema_id": model_schema_id, |
| 22 | "provider_model_id": provider_model_id, |
| 23 | "model_type": model_type, |
| 24 | "properties": properties, |
| 25 | } |
| 26 | if credentials is not None: |
| 27 | payload["credentials"] = credentials |
| 28 | else: |
| 29 | payload["encrypted_credentials"] = encrypted_credentials |
| 30 | |
| 31 | async with aiohttp.ClientSession() as session: |
| 32 | response = await session.post( |
| 33 | f"{CONFIG.TASKINGAI_INFERENCE_URL}/v1/verify_credentials", |
| 34 | json=payload, |
| 35 | ) |
| 36 | return ResponseWrapper(response.status, await response.json()) |
| 37 | |
| 38 | |
| 39 | async def health_check() -> ResponseWrapper: |
no test coverage detected