Method add_artifact

tests/unittests/cli/test_fast_api.py:314–354

Synchronous helper for tests to add artifacts.

(
        self,
        *,
        app_name: str,
        user_id: str,
        session_id: str,
        filename: str,
        artifact: types.Part,
        custom_metadata: Optional[dict[str, Any]] = None,
        canonical_uri: Optional[str] = None,
        mime_type: Optional[str] = None,
    )

      return version

    def add_artifact(
        self,
        *,
        app_name: str,
        user_id: str,
        session_id: str,
        filename: str,
        artifact: types.Part,
        custom_metadata: Optional[dict[str, Any]] = None,
        canonical_uri: Optional[str] = None,
        mime_type: Optional[str] = None,
    ) -> int:
      """Synchronous helper for tests to add artifacts."""
      key = _artifact_key(app_name, user_id, session_id, filename)
      entries = artifacts.setdefault(key, [])
      version = len(entries)
      artifact_version = ArtifactVersion(
          version=version,
          canonical_uri=(
              canonical_uri
              or _canonical_uri(
                  app_name, user_id, session_id, filename, version
              )
          ),
          custom_metadata=custom_metadata or {},
      )
      if mime_type:
        artifact_version.mime_type = mime_type
      elif artifact.inline_data is not None:
        artifact_version.mime_type = artifact.inline_data.mime_type
      elif artifact.text is not None:
        artifact_version.mime_type = "text/plain"
      elif artifact.file_data is not None:
        artifact_version.mime_type = artifact.file_data.mime_type

      entries.append({
          "version": version,
          "artifact": artifact,
          "metadata": artifact_version,
      })
      return version

    async def load_artifact(
        self, app_name, user_id, session_id, filename, version=None