src/baml_client/sync_client.py

# ----------------------------------------------------------------------------
#
#  Welcome to Baml! To use this generated code, please run the following:
#
#  $ pip install baml
#
# ----------------------------------------------------------------------------

# This file was generated by BAML: please do not edit it. Instead, edit the
# BAML files and re-generate this code using: baml-cli generate
# baml-cli is available with the baml package.

import typing
import typing_extensions
import baml_py

from . import stream_types, types, type_builder
from .parser import LlmResponseParser, LlmStreamParser
from .runtime import DoNotUseDirectlyCallManager, BamlCallOptions
from .globals import DO_NOT_USE_DIRECTLY_UNLESS_YOU_KNOW_WHAT_YOURE_DOING_RUNTIME as __runtime__

class BamlSyncClient:
    __options: DoNotUseDirectlyCallManager
    __stream_client: "BamlStreamClient"
    __http_request: "BamlHttpRequestClient"
    __http_stream_request: "BamlHttpStreamRequestClient"
    __llm_response_parser: LlmResponseParser
    __llm_stream_parser: LlmStreamParser

    def __init__(self, options: DoNotUseDirectlyCallManager):
        self.__options = options
        self.__stream_client = BamlStreamClient(options)
        self.__http_request = BamlHttpRequestClient(options)
        self.__http_stream_request = BamlHttpStreamRequestClient(options)
        self.__llm_response_parser = LlmResponseParser(options)
        self.__llm_stream_parser = LlmStreamParser(options)

    def __getstate__(self):
        # Return state needed for pickling
        return {"options": self.__options}

    def __setstate__(self, state):
        # Restore state from pickling
        self.__options = state["options"]
        self.__stream_client = BamlStreamClient(self.__options)
        self.__http_request = BamlHttpRequestClient(self.__options)
        self.__http_stream_request = BamlHttpStreamRequestClient(self.__options)
        self.__llm_response_parser = LlmResponseParser(self.__options)
        self.__llm_stream_parser = LlmStreamParser(self.__options)

    def with_options(self,
        tb: typing.Optional[type_builder.TypeBuilder] = None,
        client_registry: typing.Optional[baml_py.baml_py.ClientRegistry] = None,
        client: typing.Optional[str] = None,
        collector: typing.Optional[typing.Union[baml_py.baml_py.Collector, typing.List[baml_py.baml_py.Collector]]] = None,
        env: typing.Optional[typing.Dict[str, typing.Optional[str]]] = None,
        tags: typing.Optional[typing.Dict[str, str]] = None,
        on_tick: typing.Optional[typing.Callable[[str, baml_py.baml_py.FunctionLog], None]] = None,
    ) -> "BamlSyncClient":
        options: BamlCallOptions = {}
        if tb is not None:
            options["tb"] = tb
        if client_registry is not None:
            options["client_registry"] = client_registry
        if client is not None:
            options["client"] = client
        if collector is not None:
            options["collector"] = collector
        if env is not None:
            options["env"] = env
        if tags is not None:
            options["tags"] = tags
        if on_tick is not None:
            options["on_tick"] = on_tick
        return BamlSyncClient(self.__options.merge_options(options))

    @property
    def stream(self):
      return self.__stream_client

    @property
    def request(self):
      return self.__http_request

    @property
    def stream_request(self):
      return self.__http_stream_request

    @property
    def parse(self):
      return self.__llm_response_parser

    @property
    def parse_stream(self):
      return self.__llm_stream_parser

    def FilterItems(self, items: typing.List["types.FilterItem"],
        baml_options: BamlCallOptions = {},
    ) -> typing.List["types.Verdict"]:
        # Check if on_tick is provided
        if 'on_tick' in baml_options:
            __stream__ = self.stream.FilterItems(items=items,
                baml_options=baml_options)
            return __stream__.get_final_response()
        else:
            # Original non-streaming code
            __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="FilterItems", args={
                "items": items,
            })
            return typing.cast(typing.List["types.Verdict"], __result__.cast_to(types, types, stream_types, False, __runtime__))
    def FilterScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> typing.List["types.RankedItem"]:
        # Check if on_tick is provided
        if 'on_tick' in baml_options:
            __stream__ = self.stream.FilterScoreItems(items=items,
                baml_options=baml_options)
            return __stream__.get_final_response()
        else:
            # Original non-streaming code
            __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="FilterScoreItems", args={
                "items": items,
            })
            return typing.cast(typing.List["types.RankedItem"], __result__.cast_to(types, types, stream_types, False, __runtime__))
    def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"],
        baml_options: BamlCallOptions = {},
    ) -> types.EpisodeOutline:
        # Check if on_tick is provided
        if 'on_tick' in baml_options:
            __stream__ = self.stream.GenerateOutline(date=date,stories=stories,
                baml_options=baml_options)
            return __stream__.get_final_response()
        else:
            # Original non-streaming code
            __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="GenerateOutline", args={
                "date": date,"stories": stories,
            })
            return typing.cast(types.EpisodeOutline, __result__.cast_to(types, types, stream_types, False, __runtime__))
    def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None,
        baml_options: BamlCallOptions = {},
    ) -> typing.List["types.ScriptLine"]:
        # Check if on_tick is provided
        if 'on_tick' in baml_options:
            __stream__ = self.stream.GenerateScriptSegment(date=date,story_title=story_title,content_chunk=content_chunk,previous_context=previous_context,segment_type=segment_type,outline_context=outline_context,
                baml_options=baml_options)
            return __stream__.get_final_response()
        else:
            # Original non-streaming code
            __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="GenerateScriptSegment", args={
                "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context,
            })
            return typing.cast(typing.List["types.ScriptLine"], __result__.cast_to(types, types, stream_types, False, __runtime__))
    def ScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> typing.List["types.ScoredItem"]:
        # Check if on_tick is provided
        if 'on_tick' in baml_options:
            __stream__ = self.stream.ScoreItems(items=items,
                baml_options=baml_options)
            return __stream__.get_final_response()
        else:
            # Original non-streaming code
            __result__ = self.__options.merge_options(baml_options).call_function_sync(function_name="ScoreItems", args={
                "items": items,
            })
            return typing.cast(typing.List["types.ScoredItem"], __result__.cast_to(types, types, stream_types, False, __runtime__))
    


class BamlStreamClient:
    __options: DoNotUseDirectlyCallManager

    def __init__(self, options: DoNotUseDirectlyCallManager):
        self.__options = options

    def FilterItems(self, items: typing.List["types.FilterItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.BamlSyncStream[typing.List["stream_types.Verdict"], typing.List["types.Verdict"]]:
        __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="FilterItems", args={
            "items": items,
        })
        return baml_py.BamlSyncStream[typing.List["stream_types.Verdict"], typing.List["types.Verdict"]](
          __result__,
          lambda x: typing.cast(typing.List["stream_types.Verdict"], x.cast_to(types, types, stream_types, True, __runtime__)),
          lambda x: typing.cast(typing.List["types.Verdict"], x.cast_to(types, types, stream_types, False, __runtime__)),
          __ctx__,
        )
    def FilterScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.BamlSyncStream[typing.List["stream_types.RankedItem"], typing.List["types.RankedItem"]]:
        __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="FilterScoreItems", args={
            "items": items,
        })
        return baml_py.BamlSyncStream[typing.List["stream_types.RankedItem"], typing.List["types.RankedItem"]](
          __result__,
          lambda x: typing.cast(typing.List["stream_types.RankedItem"], x.cast_to(types, types, stream_types, True, __runtime__)),
          lambda x: typing.cast(typing.List["types.RankedItem"], x.cast_to(types, types, stream_types, False, __runtime__)),
          __ctx__,
        )
    def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.BamlSyncStream[stream_types.EpisodeOutline, types.EpisodeOutline]:
        __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="GenerateOutline", args={
            "date": date,"stories": stories,
        })
        return baml_py.BamlSyncStream[stream_types.EpisodeOutline, types.EpisodeOutline](
          __result__,
          lambda x: typing.cast(stream_types.EpisodeOutline, x.cast_to(types, types, stream_types, True, __runtime__)),
          lambda x: typing.cast(types.EpisodeOutline, x.cast_to(types, types, stream_types, False, __runtime__)),
          __ctx__,
        )
    def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None,
        baml_options: BamlCallOptions = {},
    ) -> baml_py.BamlSyncStream[typing.List["stream_types.ScriptLine"], typing.List["types.ScriptLine"]]:
        __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="GenerateScriptSegment", args={
            "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context,
        })
        return baml_py.BamlSyncStream[typing.List["stream_types.ScriptLine"], typing.List["types.ScriptLine"]](
          __result__,
          lambda x: typing.cast(typing.List["stream_types.ScriptLine"], x.cast_to(types, types, stream_types, True, __runtime__)),
          lambda x: typing.cast(typing.List["types.ScriptLine"], x.cast_to(types, types, stream_types, False, __runtime__)),
          __ctx__,
        )
    def ScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.BamlSyncStream[typing.List["stream_types.ScoredItem"], typing.List["types.ScoredItem"]]:
        __ctx__, __result__ = self.__options.merge_options(baml_options).create_sync_stream(function_name="ScoreItems", args={
            "items": items,
        })
        return baml_py.BamlSyncStream[typing.List["stream_types.ScoredItem"], typing.List["types.ScoredItem"]](
          __result__,
          lambda x: typing.cast(typing.List["stream_types.ScoredItem"], x.cast_to(types, types, stream_types, True, __runtime__)),
          lambda x: typing.cast(typing.List["types.ScoredItem"], x.cast_to(types, types, stream_types, False, __runtime__)),
          __ctx__,
        )
    

class BamlHttpRequestClient:
    __options: DoNotUseDirectlyCallManager

    def __init__(self, options: DoNotUseDirectlyCallManager):
        self.__options = options

    def FilterItems(self, items: typing.List["types.FilterItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterItems", args={
            "items": items,
        }, mode="request")
        return __result__
    def FilterScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterScoreItems", args={
            "items": items,
        }, mode="request")
        return __result__
    def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateOutline", args={
            "date": date,"stories": stories,
        }, mode="request")
        return __result__
    def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None,
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateScriptSegment", args={
            "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context,
        }, mode="request")
        return __result__
    def ScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="ScoreItems", args={
            "items": items,
        }, mode="request")
        return __result__
    

class BamlHttpStreamRequestClient:
    __options: DoNotUseDirectlyCallManager

    def __init__(self, options: DoNotUseDirectlyCallManager):
        self.__options = options

    def FilterItems(self, items: typing.List["types.FilterItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterItems", args={
            "items": items,
        }, mode="stream")
        return __result__
    def FilterScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="FilterScoreItems", args={
            "items": items,
        }, mode="stream")
        return __result__
    def GenerateOutline(self, date: str,stories: typing.List["types.StoryItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateOutline", args={
            "date": date,"stories": stories,
        }, mode="stream")
        return __result__
    def GenerateScriptSegment(self, date: str,story_title: str,content_chunk: str,previous_context: typing.Optional[str],segment_type: str,outline_context: typing.Optional[str] = None,
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="GenerateScriptSegment", args={
            "date": date,"story_title": story_title,"content_chunk": content_chunk,"previous_context": previous_context,"segment_type": segment_type,"outline_context": outline_context,
        }, mode="stream")
        return __result__
    def ScoreItems(self, items: typing.List["types.ScoreItem"],
        baml_options: BamlCallOptions = {},
    ) -> baml_py.baml_py.HTTPRequest:
        __result__ = self.__options.merge_options(baml_options).create_http_request_sync(function_name="ScoreItems", args={
            "items": items,
        }, mode="stream")
        return __result__
    

b = BamlSyncClient(DoNotUseDirectlyCallManager({}))