Coverage for langsmith/_internal/_multipart.py: 0%

1from __future__ import annotations 

2 

3from collections.abc import Iterable 

4from io import BufferedReader 

5from typing import Union 

6 

7MultipartPart = tuple[ 

8 str, tuple[None, Union[bytes, BufferedReader], str, dict[str, str]] 

9] 

10 

11 

12class MultipartPartsAndContext: 

13 parts: list[MultipartPart] 

14 context: str 

15 

16 __slots__ = ("parts", "context") 

17 

18 def __init__(self, parts: list[MultipartPart], context: str) -> None: 

19 self.parts = parts 

20 self.context = context 

21 

22 

23def join_multipart_parts_and_context( 

24 parts_and_contexts: Iterable[MultipartPartsAndContext], 

25) -> MultipartPartsAndContext: 

26 acc_parts: list[MultipartPart] = [] 

27 acc_context: list[str] = [] 

28 for parts_and_context in parts_and_contexts: 

29 acc_parts.extend(parts_and_context.parts) 

30 acc_context.append(parts_and_context.context) 

31 return MultipartPartsAndContext(acc_parts, "; ".join(acc_context))