scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

chore(typing): replace Any kwargs unpacking with validated model parsing

Browse source 
Swap `Model(**payload)` for `Model.model_validate(payload)` at the seams where
the payload comes back untyped, so basedpyright stops widening every target
field to Any. None of the models involved override `__init__`, so validation
goes through the same core validator either way.

Also route UserRepository through its own typed helpers (find_many, update,
find_by_id) instead of the raw Prisma table, drop the redundant `_to_model`
override signature, and call generate_key_helper_fn with explicit arguments in
the SSO callback rather than splatting an untyped dict.

Whole-tree basedpyright: reportAny 21481 -> 20834, reportExplicitAny 7258 ->
7252, with every other rule unchanged or lower.
This commit is contained in:
mateo-berri 2026-07-31 19:14:42 +00:00
parent f54f92437b
commit c0cab45350
No known key found for this signature in database
14 changed files with 76 additions and 84 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
basedpyright-code-budget.json
View file

@ -1,6 +1,6 @@
{
"reportAny": {
"limit": 31903
"limit": 31256
},
"reportArgumentType": {
"limit": 2645
@ -24,7 +24,7 @@
"limit": 42
},
"reportExplicitAny": {
"limit": 10214
"limit": 10208
},
"reportFunctionMemberAccess": {
"limit": 11
@ -33,7 +33,7 @@
"limit": 227
},
"reportIncompatibleMethodOverride": {
"limit": 78
"limit": 77
},
"reportIncompatibleVariableOverride": {
"limit": 12
@ -99,7 +99,7 @@
"limit": 0
},
"reportUnknownArgumentType": {
"limit": 45366
"limit": 45357
},
"reportUnknownLambdaType": {
"limit": 113

8
enterprise/litellm_enterprise/proxy/hooks/managed_files.py
View file

@ -215,7 +215,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
)
if result:
return LiteLLM_ManagedFileTable(**result)
return LiteLLM_ManagedFileTable.model_validate(result)
## CHECK DB
db_object = await self.prisma_client.db.litellm_managedfiletable.find_first(
@ -223,7 +223,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
)
if db_object:
return LiteLLM_ManagedFileTable(**db_object.model_dump())
return LiteLLM_ManagedFileTable.model_validate(db_object.model_dump())
return None
async def delete_unified_file_id(
@ -349,7 +349,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
if isinstance(batch.file_object, str)
else batch.file_object
)
batch_obj = LiteLLMBatch(**batch_data)
batch_obj = LiteLLMBatch.model_validate(batch_data)
batch_obj.id = batch.unified_object_id
batch_objects.append(batch_obj)
@ -382,7 +382,7 @@ class _PROXY_LiteLLMManagedFiles(CustomLogger, BaseFileEndpoints):
"flat_model_file_ids": {"hasSome": model_object_ids},
}
)
return [OpenAIFileObject(**file_object.file_object) for file_object in file_ids]
return [OpenAIFileObject.model_validate(file_object.file_object) for file_object in file_ids]
async def check_managed_file_id_access(
self, data: Dict, user_api_key_dict: UserAPIKeyAuth

4
litellm/litellm_core_utils/streaming_handler.py
View file

@ -131,8 +131,8 @@ class CustomStreamWrapper:
self.sent_last_chunk = False
self._stream_created_time: float = time.time()
litellm_params: GenericLiteLLMParams = GenericLiteLLMParams(
**self.logging_obj.model_call_details.get("litellm_params", {})
litellm_params: GenericLiteLLMParams = GenericLiteLLMParams.model_validate(
dict(**self.logging_obj.model_call_details.get("litellm_params", {}))
)
self.merge_reasoning_content_in_choices: bool = litellm_params.merge_reasoning_content_in_choices or False
self.sent_first_thinking_block = False

4
litellm/llms/azure/responses/transformation.py
View file

@ -66,7 +66,7 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
)
# Create ResponseReasoningItem object from the item data
reasoning_item = ResponseReasoningItem(**item_data)
reasoning_item = ResponseReasoningItem.model_validate(item_data)
# Convert back to dict with exclude_none=True to exclude None fields
dict_reasoning_item = reasoning_item.model_dump(exclude_none=True)
@ -346,4 +346,4 @@ class AzureOpenAIResponsesAPIConfig(OpenAIResponsesAPIConfig):
from litellm.llms.azure.chat.gpt_transformation import AzureOpenAIError
raise AzureOpenAIError(message=raw_response.text, status_code=raw_response.status_code)
return ResponsesAPIResponse(**raw_response_json)
return ResponsesAPIResponse.model_validate(raw_response_json)

10
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -6184,10 +6184,12 @@ class BaseLLMHTTPHandler:
import websockets
from websockets.asyncio.client import ClientConnection
litellm_params = GenericLiteLLMParams(
api_base=api_base,
api_key=api_key,
**kwargs,
litellm_params = GenericLiteLLMParams.model_validate(
{
"api_base": api_base,
"api_key": api_key,
**kwargs,
}
)
headers = responses_api_provider_config.validate_environment(
headers={},

4
litellm/llms/manus/responses/transformation.py
View file

@ -217,7 +217,7 @@ class ManusResponsesAPIConfig(OpenAIResponsesAPIConfig):
raw_response_json["id"] = f"unknown-{uuid.uuid4().hex[:8]}"
try:
response = ResponsesAPIResponse(**raw_response_json)
response = ResponsesAPIResponse.model_validate(raw_response_json)
except Exception:
verbose_logger.debug(f"Error constructing ResponsesAPIResponse: {raw_response_json}, using model_construct")
response = ResponsesAPIResponse.model_construct(**raw_response_json)
@ -305,7 +305,7 @@ class ManusResponsesAPIConfig(OpenAIResponsesAPIConfig):
raw_response_json["id"] = f"unknown-{uuid.uuid4().hex[:8]}"
try:
response = ResponsesAPIResponse(**raw_response_json)
response = ResponsesAPIResponse.model_validate(raw_response_json)
except Exception:
verbose_logger.debug(f"Error constructing ResponsesAPIResponse: {raw_response_json}, using model_construct")
response = ResponsesAPIResponse.model_construct(**raw_response_json)

13
litellm/proxy/hooks/user_management_event_hooks.py
View file

@ -6,8 +6,6 @@ import asyncio
from datetime import datetime, timezone
from typing import Optional
from pydantic import BaseModel
import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
@ -16,7 +14,6 @@ from litellm.proxy._types import (
CommonProxyErrors,
LiteLLM_AuditLogs,
Litellm_EntityType,
LiteLLM_UserTable,
LitellmTableNames,
NewUserRequest,
NewUserResponse,
@ -58,11 +55,11 @@ class UserManagementEventHooks:
try:
if prisma_client is None:
raise Exception(CommonProxyErrors.db_not_connected_error.value)
user_row: BaseModel = await UserRepository(prisma_client).table.find_first(
where={"user_id": response.user_id}
)
user_row_litellm_typed = LiteLLM_UserTable(**user_row.model_dump(exclude_none=True))
if response.user_id is None:
raise Exception("no user_id returned for the newly created user")
user_row_litellm_typed = await UserRepository(prisma_client).find_by_id(response.user_id)
if user_row_litellm_typed is None:
raise Exception(f"no user row found for user_id={response.user_id}")
asyncio.create_task(
UserManagementEventHooks.create_internal_user_audit_log(
user_id=user_row_litellm_typed.user_id,

26
litellm/proxy/management_endpoints/ui_sso.py
View file

@ -3137,7 +3137,7 @@ class SSOAuthenticationHandler:
_default_team_params = deepcopy(litellm.default_team_params)
_new_team_request = team_request.model_dump()
_new_team_request.update(_default_team_params)
team_request = NewTeamRequest(**_new_team_request)
team_request = NewTeamRequest.model_validate(_new_team_request)
return team_request
@staticmethod
@ -3271,14 +3271,6 @@ class SSOAuthenticationHandler:
# User might not be already created on first generation of key
# But if it is, we want their models preferences
default_ui_key_values: Dict[str, Any] = {
"duration": LITELLM_UI_SESSION_DURATION,
"key_max_budget": litellm.max_ui_session_budget,
"aliases": {},
"config": {},
"spend": 0,
"team_id": "litellm-dashboard",
}
user_defined_values: Optional[SSOUserDefinedValues] = None
if user_custom_sso is not None:
@ -3338,10 +3330,20 @@ class SSOAuthenticationHandler:
verbose_proxy_logger.info(f"user_defined_values for creating ui key: {user_defined_values}")
default_ui_key_values.update(user_defined_values)
default_ui_key_values["request_type"] = "key"
response = await generate_key_helper_fn(
**default_ui_key_values, # type: ignore
request_type="key",
duration=LITELLM_UI_SESSION_DURATION,
key_max_budget=litellm.max_ui_session_budget,
aliases={},
config={},
spend=0,
team_id="litellm-dashboard",
models=user_defined_values["models"],
user_id=user_defined_values["user_id"],
user_email=user_defined_values["user_email"],
user_role=user_defined_values["user_role"],
max_budget=user_defined_values["max_budget"],
budget_duration=user_defined_values["budget_duration"],
table_name="key",
)

2
litellm/proxy/openai_files_endpoints/common_utils.py
View file

@ -1035,7 +1035,7 @@ async def get_batch_from_database(
if isinstance(db_batch_object.file_object, str)
else db_batch_object.file_object
)
response = LiteLLMBatch(**batch_data)
response = LiteLLMBatch.model_validate(batch_data)
response.id = batch_id
# The stored batch object has the raw provider input_file_id. Resolve to unified ID.

57
litellm/repositories/user_repository.py
View file

@ -3,61 +3,56 @@ User repository for database operations on LiteLLM_UserTable.
"""
import json
from typing import Any, Dict, List, Optional, Type
from typing import Any, Dict, List, Mapping, Optional, Type
from litellm.models.user import LiteLLM_UserTable
from litellm.repositories.base_repository import BaseRepository
from litellm.repositories.base_repository import BaseRepository, DbRecord, record_to_dict
_JSON_ENCODED_COLUMNS = frozenset({"metadata", "model_spend", "model_max_budget"})
class UserRepository(BaseRepository[LiteLLM_UserTable]):
"""Repository for user database operations."""
@property
def table(self) -> Any:
def table(self) -> Any: # any-ok: Prisma table actions are reached through the untyped client wrapper
return self.prisma_client.db.litellm_usertable
@property
def model_class(self) -> Type[LiteLLM_UserTable]:
return LiteLLM_UserTable
def _to_model(self, record: Any) -> Optional[LiteLLM_UserTable]:
def _to_model(self, record: Optional[DbRecord]) -> Optional[LiteLLM_UserTable]:
"""Convert a database record to a User model."""
if record is None:
return None
data = record.dict() if hasattr(record, "dict") else dict(record)
return LiteLLM_UserTable.model_validate(
{
column: json.loads(value) if column in _JSON_ENCODED_COLUMNS and isinstance(value, str) else value
for column, value in record_to_dict(record).items()
}
)
json_fields = ["metadata", "model_spend", "model_max_budget"]
for field in json_fields:
if isinstance(data.get(field), str):
data[field] = json.loads(data[field])
return LiteLLM_UserTable(**data)
async def find_by_id(self, user_id: str, id_field: str = "user_id") -> Optional[LiteLLM_UserTable]:
return await super().find_by_id(user_id, id_field)
async def find_by_id(self, id_value: str, id_field: str = "user_id") -> Optional[LiteLLM_UserTable]:
return await super().find_by_id(id_value, id_field)
async def find_by_email(self, user_email: str) -> Optional[LiteLLM_UserTable]:
"""Find a user by email."""
records = await self.table.find_many(where={"user_email": user_email})
if records:
return self._to_model(records[0])
return None
records = await self.find_many(where={"user_email": user_email})
return records[0] if records else None
async def find_by_sso_id(self, sso_user_id: str) -> Optional[LiteLLM_UserTable]:
"""Find a user by SSO ID."""
record = await self.table.find_unique(where={"sso_user_id": sso_user_id})
return self._to_model(record)
return await self.find_by_id(sso_user_id, id_field="sso_user_id")
async def find_by_organization_id(self, organization_id: str) -> List[LiteLLM_UserTable]:
"""Find all users in an organization."""
records = await self.table.find_many(where={"organization_id": organization_id})
return self._to_model_list(records)
return await self.find_many(where={"organization_id": organization_id})
async def find_by_team_id(self, team_id: str) -> List[LiteLLM_UserTable]:
"""Find all users in a team."""
records = await self.table.find_many(where={"teams": {"has": team_id}})
return self._to_model_list(records)
return await self.find_many(where={"teams": {"has": team_id}})
async def count_billable_users(self) -> int:
"""Number of users that count toward the license seat limit.
@ -86,7 +81,7 @@ class UserRepository(BaseRepository[LiteLLM_UserTable]):
max_budget: Optional[float] = None,
user_email: Optional[str] = None,
models: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
metadata: Optional[Mapping[str, object]] = None,
max_parallel_requests: Optional[int] = None,
tpm_limit: Optional[int] = None,
rpm_limit: Optional[int] = None,
@ -96,7 +91,7 @@ class UserRepository(BaseRepository[LiteLLM_UserTable]):
object_permission_id: Optional[str] = None,
) -> LiteLLM_UserTable:
"""Create a new user."""
data: Dict[str, Any] = {"user_id": user_id}
data: Dict[str, object] = {"user_id": user_id}
if user_alias is not None:
data["user_alias"] = user_alias
if team_id is not None:
@ -149,7 +144,7 @@ class UserRepository(BaseRepository[LiteLLM_UserTable]):
max_budget: Optional[float] = None,
user_email: Optional[str] = None,
models: Optional[List[str]] = None,
metadata: Optional[Dict[str, Any]] = None,
metadata: Optional[Mapping[str, object]] = None,
max_parallel_requests: Optional[int] = None,
tpm_limit: Optional[int] = None,
rpm_limit: Optional[int] = None,
@ -159,7 +154,7 @@ class UserRepository(BaseRepository[LiteLLM_UserTable]):
object_permission_id: Optional[str] = None,
) -> Optional[LiteLLM_UserTable]:
"""Update a user."""
data: Dict[str, Any] = {}
data: Dict[str, object] = {}
if user_alias is not None:
data["user_alias"] = user_alias
if team_id is not None:
@ -212,11 +207,7 @@ class UserRepository(BaseRepository[LiteLLM_UserTable]):
if not await self.exists(user_id, id_field="user_id"):
return None
record = await self.table.update(
where={"user_id": user_id},
data={"teams": {"push": team_id}},
)
return self._to_model(record)
return await self.update(user_id, {"teams": {"push": team_id}}, id_field="user_id")
async def remove_from_team(self, user_id: str, team_id: str) -> Optional[LiteLLM_UserTable]:
"""Remove a user from a team.

12
litellm/router.py
View file

@ -8613,9 +8613,9 @@ class Router:
deployment = self.get_deployment(model_id=model_id)
if deployment is None or self._is_deployment_blocked(deployment):
return None
return CredentialLiteLLMParams(**deployment.litellm_params.model_dump(exclude_none=True)).model_dump(
exclude_none=True
)
return CredentialLiteLLMParams.model_validate(
deployment.litellm_params.model_dump(exclude_none=True)
).model_dump(exclude_none=True)
def get_deployment_by_model_group_name(self, model_group_name: str) -> Optional[Deployment]:
"""
@ -8755,9 +8755,9 @@ class Router:
return None
# Get basic credentials
credentials = CredentialLiteLLMParams(**deployment.litellm_params.model_dump(exclude_none=True)).model_dump(
exclude_none=True
)
credentials = CredentialLiteLLMParams.model_validate(
deployment.litellm_params.model_dump(exclude_none=True)
).model_dump(exclude_none=True)
# Resolve litellm_credential_name to actual credentials
if deployment.litellm_params.litellm_credential_name is not None:

2
litellm/router_strategy/complexity_router/complexity_router.py
View file

@ -329,7 +329,7 @@ class ComplexityRouter(CustomLogger):
# Parse config - always create a new instance to avoid singleton mutation
if complexity_router_config:
self.config = ComplexityRouterConfig(**complexity_router_config)
self.config = ComplexityRouterConfig.model_validate(complexity_router_config)
else:
self.config = ComplexityRouterConfig()

4
ruff-strict-budget.json
View file

@ -24,7 +24,7 @@
"limit": 130
},
"ANN401": {
"limit": 2010
"limit": 2009
},
"ASYNC230": {
"limit": 14
@ -324,7 +324,7 @@
"limit": 879
},
"UP006": {
"limit": 12138
"limit": 12135
},
"UP007": {
"limit": 2526

6
type-discipline-budget.json
View file

@ -1,9 +1,9 @@
{
"LIT001": {
"limit": 23253
"limit": 23250
},
"LIT002": {
"limit": 27427
"limit": 27425
},
"LIT003": {
"limit": 292
@ -24,6 +24,6 @@
"limit": 1004
},
"LIT009": {
"limit": 2474
"limit": 2473
}
}