fix: replace mutable default arguments (= [] and = {}) with None

Mutable default arguments in Python are shared across all calls,
causing subtle state-leak bugs. Replaced with Optional[...] = None
and initialized inside the function body.

Affected:
- utils.decode(): tokens: List[int] = [] -> Optional[List[int]] = None
- VectorStoreRegistry.__init__(): vector_stores = [] -> None
- get_litellm_model_info(): model: dict = {} -> Optional[dict] = None
- LowestCostLoggingHandler.__init__(): routing_args = {} -> None
- LowestLatencyLoggingHandler.__init__(): routing_args = {} -> None
- LowestTPM_RPM_Routing.__init__(): routing_args = {} -> None
- LowestTPM_RPM_Routing_v2.__init__(): routing_args = {} -> None
This commit is contained in:
zach 2026-04-10 22:25:46 -07:00
parent 4e12d3c562
commit 2df755425a
7 changed files with 18 additions and 8 deletions

View file

@ -5954,7 +5954,9 @@ def select_data_generator(
)
def get_litellm_model_info(model: dict = {}):
def get_litellm_model_info(model: Optional[dict] = None):
if model is None:
model = {}
model_info = model.get("model_info", {})
model_to_lookup = model.get("litellm_params", {}).get("model", None)
try:

View file

@ -15,7 +15,7 @@ class LowestCostLoggingHandler(CustomLogger):
logged_success: int = 0
logged_failure: int = 0
def __init__(self, router_cache: DualCache, routing_args: dict = {}):
def __init__(self, router_cache: DualCache, routing_args: Optional[dict] = None):
self.router_cache = router_cache
def log_success_event(self, kwargs, response_obj, start_time, end_time):

View file

@ -31,8 +31,10 @@ class LowestLatencyLoggingHandler(CustomLogger):
logged_success: int = 0
logged_failure: int = 0
def __init__(self, router_cache: DualCache, routing_args: dict = {}):
def __init__(self, router_cache: DualCache, routing_args: Optional[dict] = None):
self.router_cache = router_cache
if routing_args is None:
routing_args = {}
self.routing_args = RoutingArgs(**routing_args)
def log_success_event( # noqa: PLR0915

View file

@ -22,8 +22,10 @@ class LowestTPMLoggingHandler(CustomLogger):
logged_failure: int = 0
default_cache_time_seconds: int = 1 * 60 * 60 # 1 hour
def __init__(self, router_cache: DualCache, routing_args: dict = {}):
def __init__(self, router_cache: DualCache, routing_args: Optional[dict] = None):
self.router_cache = router_cache
if routing_args is None:
routing_args = {}
self.routing_args = RoutingArgs(**routing_args)
def log_success_event(self, kwargs, response_obj, start_time, end_time):

View file

@ -47,8 +47,10 @@ class LowestTPMLoggingHandler_v2(BaseRoutingStrategy, CustomLogger):
logged_failure: int = 0
default_cache_time_seconds: int = 1 * 60 * 60 # 1 hour
def __init__(self, router_cache: DualCache, routing_args: dict = {}):
def __init__(self, router_cache: DualCache, routing_args: Optional[dict] = None):
self.router_cache = router_cache
if routing_args is None:
routing_args = {}
self.routing_args = RoutingArgs(**routing_args)
BaseRoutingStrategy.__init__(
self,

View file

@ -2236,7 +2236,9 @@ def encode(model="", text="", custom_tokenizer: Optional[dict] = None):
return enc
def decode(model="", tokens: List[int] = [], custom_tokenizer: Optional[dict] = None):
def decode(model="", tokens: Optional[List[int]] = None, custom_tokenizer: Optional[dict] = None):
if tokens is None:
tokens = []
tokenizer_json = custom_tokenizer or _select_tokenizer(model=model)
dec = tokenizer_json["tokenizer"].decode(tokens)
return dec

View file

@ -106,8 +106,8 @@ class VectorStoreIndexRegistry:
class VectorStoreRegistry:
def __init__(self, vector_stores: List[LiteLLM_ManagedVectorStore] = []):
self.vector_stores: List[LiteLLM_ManagedVectorStore] = vector_stores
def __init__(self, vector_stores: Optional[List[LiteLLM_ManagedVectorStore]] = None):
self.vector_stores: List[LiteLLM_ManagedVectorStore] = vector_stores or []
self.vector_store_ids_to_vector_store_map: Dict[
str, LiteLLM_ManagedVectorStore
] = {}