_parse() — anthropic-sdk-python Function Reference

Source Code

src/anthropic/_legacy_response.py lines 199–361

    def _parse(self, *, to: type[_T] | None = None) -> R | _T:
        cast_to = to if to is not None else self._cast_to

        # unwrap `TypeAlias('Name', T)` -> `T`
        if is_type_alias_type(cast_to):
            cast_to = cast_to.__value__  # type: ignore[unreachable]

        # unwrap `Annotated[T, ...]` -> `T`
        if cast_to and is_annotated_type(cast_to):
            cast_to = extract_type_arg(cast_to, 0)

        origin = get_origin(cast_to) or cast_to

        if inspect.isclass(origin):
            if issubclass(cast(Any, origin), JSONLDecoder):
                return cast(
                    R,
                    cast("type[JSONLDecoder[Any]]", cast_to)(
                        raw_iterator=self.http_response.iter_bytes(chunk_size=64),
                        line_type=extract_type_arg(cast_to, 0),
                        http_response=self.http_response,
                    ),
                )

            if issubclass(cast(Any, origin), AsyncJSONLDecoder):
                return cast(
                    R,
                    cast("type[AsyncJSONLDecoder[Any]]", cast_to)(
                        raw_iterator=self.http_response.aiter_bytes(chunk_size=64),
                        line_type=extract_type_arg(cast_to, 0),
                        http_response=self.http_response,
                    ),
                )

        if self._stream:
            if to:
                if not is_stream_class_type(to):
                    raise TypeError(f"Expected custom parse type to be a subclass of {Stream} or {AsyncStream}")

                return cast(
                    _T,
                    to(
                        cast_to=extract_stream_chunk_type(
                            to,
                            failure_message="Expected custom stream type to be passed with a type argument, e.g. Stream[ChunkType]",
                        ),
                        response=self.http_response,
                        client=cast(Any, self._client),
                    ),
                )

            if self._stream_cls:
                return cast(
                    R,
                    self._stream_cls(
                        cast_to=extract_stream_chunk_type(self._stream_cls),
                        response=self.http_response,
                        client=cast(Any, self._client),
                    ),
                )

            stream_cls = cast("type[Stream[Any]] | type[AsyncStream[Any]] | None", self._client._default_stream_cls)
            if stream_cls is None:
                raise MissingStreamClassError()

            return cast(
                R,
                stream_cls(
                    cast_to=cast_to,
                    response=self.http_response,
                    client=cast(Any, self._client),
                ),
            )

        if cast_to is NoneType:
            return cast(R, None)

        response = self.http_response
        if cast_to == str:
            return cast(R, response.text)