(
cls,
response_schema: Optional[Dict[str, Any]],
)
| 5159 | |
| 5160 | @classmethod |
| 5161 | def _compile_stream_plan( |
| 5162 | cls, |
| 5163 | response_schema: Optional[Dict[str, Any]], |
| 5164 | ) -> Optional[Dict[str, Any]]: |
| 5165 | if not isinstance(response_schema, dict): |
| 5166 | return None |
| 5167 | if response_schema.get("x-parser") == "json": |
| 5168 | return {"kind": "json-root"} |
| 5169 | segment_plan = cls._compile_segment_message_plan(response_schema) |
| 5170 | if segment_plan is not None: |
| 5171 | return segment_plan |
| 5172 | return cls._compile_tagged_message_plan(response_schema) |
| 5173 | |
| 5174 | @classmethod |
| 5175 | def _cached_stream_plan( |
no test coverage detected