Spaces:
Sleeping
Sleeping
File size: 1,425 Bytes
bf4ab00 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 |
from typing import TypeVar
from aworld.trace.base import Carrier
from aworld.logs.util import logger
T = TypeVar("T")
class ListTupleCarrier(Carrier):
def __init__(self, headers: list[tuple[str, T]]):
self.headers = headers
def get(self, key: str) -> T:
for header, value in self.headers:
header_str = header.decode(
'utf-8') if isinstance(header, bytes) else header
key_str = key.decode('utf-8') if isinstance(key, bytes) else key
if header_str.lower() == key_str.lower():
return value.decode('utf-8') if isinstance(value, bytes) else value
return None
def set(self, key: str, value: T) -> None:
for i, (header, _) in enumerate(self.headers):
header_str = header.decode(
'utf-8') if isinstance(header, bytes) else header
key_str = key.decode('utf-8') if isinstance(key, bytes) else key
if header_str.lower() == key_str.lower():
self.headers[i] = (header, value)
return
self.headers.append((key, value))
class DictCarrier(Carrier):
def __init__(self, headers: dict[str, T]):
self.headers = headers
def get(self, key: str) -> T:
return self.headers.get(key)
def set(self, key: str, value: T) -> None:
logger.info(f"set header {key}={value}")
self.headers[key] = value
|