(
self,
*,
app_name: str,
user_id: str,
filename: str,
artifact: types.Part,
session_id: Optional[str] = None,
custom_metadata: Optional[dict[str, Any]] = None,
)
| 271 | self.save_artifact_side_effect: Optional[BaseException] = None |
| 272 | |
| 273 | async def save_artifact( |
| 274 | self, |
| 275 | *, |
| 276 | app_name: str, |
| 277 | user_id: str, |
| 278 | filename: str, |
| 279 | artifact: types.Part, |
| 280 | session_id: Optional[str] = None, |
| 281 | custom_metadata: Optional[dict[str, Any]] = None, |
| 282 | ) -> int: |
| 283 | if self.save_artifact_side_effect is not None: |
| 284 | effect = self.save_artifact_side_effect |
| 285 | if isinstance(effect, BaseException): |
| 286 | raise effect |
| 287 | raise TypeError( |
| 288 | "save_artifact_side_effect must be an exception instance." |
| 289 | ) |
| 290 | key = _artifact_key(app_name, user_id, session_id, filename) |
| 291 | entries = artifacts.setdefault(key, []) |
| 292 | version = len(entries) |
| 293 | artifact_version = ArtifactVersion( |
| 294 | version=version, |
| 295 | canonical_uri=_canonical_uri( |
| 296 | app_name, user_id, session_id, filename, version |
| 297 | ), |
| 298 | custom_metadata=custom_metadata or {}, |
| 299 | ) |
| 300 | if artifact.inline_data is not None: |
| 301 | artifact_version.mime_type = artifact.inline_data.mime_type |
| 302 | elif artifact.text is not None: |
| 303 | artifact_version.mime_type = "text/plain" |
| 304 | elif artifact.file_data is not None: |
| 305 | artifact_version.mime_type = artifact.file_data.mime_type |
| 306 | |
| 307 | entries.append({ |
| 308 | "version": version, |
| 309 | "artifact": artifact, |
| 310 | "metadata": artifact_version, |
| 311 | }) |
| 312 | return version |
| 313 | |
| 314 | def add_artifact( |
| 315 | self, |
no test coverage detected