fix: Complete deprecated typing imports replacement

- Replace typing.Type with type in all utility files
- Replace typing.Dict with dict in remaining files
- Replace typing.List with list in remaining files
- Fix all undefined name errors from deprecated imports
- Ensure compatibility with Python 3.10-3.13 type checking

Co-Authored-By: João <joao@crewai.com>
This commit is contained in:
Devin AI
2025-09-04 02:17:59 +00:00
parent 045da4f030
commit 3257d2757f
9 changed files with 37 additions and 39 deletions

View File

@@ -1,6 +1,6 @@
import subprocess import subprocess
from enum import Enum from enum import Enum
from typing import List, Optional from typing import Optional
import click import click
from packaging import version from packaging import version

View File

@@ -2,7 +2,7 @@ from __future__ import annotations
import threading import threading
from contextlib import contextmanager from contextlib import contextmanager
from typing import Any, Callable, Type, TypeVar, cast from typing import Any, Callable, TypeVar, cast
from blinker import Signal from blinker import Signal
@@ -32,10 +32,10 @@ class CrewAIEventsBus:
def _initialize(self) -> None: def _initialize(self) -> None:
"""Initialize the event bus internal state""" """Initialize the event bus internal state"""
self._signal = Signal("crewai_event_bus") self._signal = Signal("crewai_event_bus")
self._handlers: dict[Type[BaseEvent], list[Callable]] = {} self._handlers: dict[type[BaseEvent], list[Callable]] = {}
def on( def on(
self, event_type: Type[EventT] self, event_type: type[EventT]
) -> Callable[[Callable[[Any, EventT], None]], Callable[[Any, EventT], None]]: ) -> Callable[[Callable[[Any, EventT], None]], Callable[[Any, EventT], None]]:
""" """
Decorator to register an event handler for a specific event type. Decorator to register an event handler for a specific event type.
@@ -82,7 +82,7 @@ class CrewAIEventsBus:
self._signal.send(source, event=event) self._signal.send(source, event=event)
def register_handler( def register_handler(
self, event_type: Type[EventTypes], handler: Callable[[Any, EventTypes], None] self, event_type: type[EventTypes], handler: Callable[[Any, EventTypes], None]
) -> None: ) -> None:
"""Register an event handler for a specific event type""" """Register an event handler for a specific event type"""
if event_type not in self._handlers: if event_type not in self._handlers:

View File

@@ -1,4 +1,4 @@
from typing import Dict, Optional, Union from typing import Optional, Union
from pydantic import BaseModel, Field from pydantic import BaseModel, Field

View File

@@ -1,7 +1,7 @@
import asyncio import asyncio
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from inspect import signature from inspect import signature
from typing import Any, Callable, Type, get_args, get_origin, Optional from typing import Any, Callable, get_args, get_origin, Optional
from pydantic import ( from pydantic import (
BaseModel, BaseModel,
@@ -34,7 +34,7 @@ class BaseTool(BaseModel, ABC):
"""Used to tell the model how/when/why to use the tool.""" """Used to tell the model how/when/why to use the tool."""
env_vars: list[EnvVar] = [] env_vars: list[EnvVar] = []
"""List of environment variables used by the tool.""" """List of environment variables used by the tool."""
args_schema: Type[PydanticBaseModel] = Field( args_schema: type[PydanticBaseModel] = Field(
default_factory=_ArgsSchemaPlaceholder, validate_default=True default_factory=_ArgsSchemaPlaceholder, validate_default=True
) )
"""The schema for the arguments that the tool accepts.""" """The schema for the arguments that the tool accepts."""
@@ -52,8 +52,8 @@ class BaseTool(BaseModel, ABC):
@field_validator("args_schema", mode="before") @field_validator("args_schema", mode="before")
@classmethod @classmethod
def _default_args_schema( def _default_args_schema(
cls, v: Type[PydanticBaseModel] cls, v: type[PydanticBaseModel]
) -> Type[PydanticBaseModel]: ) -> type[PydanticBaseModel]:
if not isinstance(v, cls._ArgsSchemaPlaceholder): if not isinstance(v, cls._ArgsSchemaPlaceholder):
return v return v

View File

@@ -1,17 +1,17 @@
from typing import Any, Type from typing import Any
from pydantic import BaseModel from pydantic import BaseModel
def process_config( def process_config(
values: dict[str, Any], model_class: Type[BaseModel] values: dict[str, Any], model_class: type[BaseModel]
) -> dict[str, Any]: ) -> dict[str, Any]:
""" """
Process the config dictionary and update the values accordingly. Process the config dictionary and update the values accordingly.
Args: Args:
values (dict[str, Any]): The dictionary of values to update. values (dict[str, Any]): The dictionary of values to update.
model_class (Type[BaseModel]): The Pydantic model class to reference for field validation. model_class (type[BaseModel]): The Pydantic model class to reference for field validation.
Returns: Returns:
dict[str, Any]: The updated values dictionary. dict[str, Any]: The updated values dictionary.

View File

@@ -1,6 +1,6 @@
import json import json
import re import re
from typing import Any, Optional, Type, Union, get_args, get_origin from typing import Any, Optional, Union, get_args, get_origin
from pydantic import BaseModel, ValidationError from pydantic import BaseModel, ValidationError
@@ -116,10 +116,10 @@ class Converter(OutputConverter):
def convert_to_model( def convert_to_model(
result: str, result: str,
output_pydantic: Optional[Type[BaseModel]], output_pydantic: Optional[type[BaseModel]],
output_json: Optional[Type[BaseModel]], output_json: Optional[type[BaseModel]],
agent: Any, agent: Any,
converter_cls: Optional[Type[Converter]] = None, converter_cls: Optional[type[Converter]] = None,
) -> Union[dict, BaseModel, str]: ) -> Union[dict, BaseModel, str]:
model = output_pydantic or output_json model = output_pydantic or output_json
if model is None: if model is None:
@@ -146,7 +146,7 @@ def convert_to_model(
def validate_model( def validate_model(
result: str, model: Type[BaseModel], is_json_output: bool result: str, model: type[BaseModel], is_json_output: bool
) -> Union[dict, BaseModel]: ) -> Union[dict, BaseModel]:
exported_result = model.model_validate_json(result) exported_result = model.model_validate_json(result)
if is_json_output: if is_json_output:
@@ -156,10 +156,10 @@ def validate_model(
def handle_partial_json( def handle_partial_json(
result: str, result: str,
model: Type[BaseModel], model: type[BaseModel],
is_json_output: bool, is_json_output: bool,
agent: Any, agent: Any,
converter_cls: Optional[Type[Converter]] = None, converter_cls: Optional[type[Converter]] = None,
) -> Union[dict, BaseModel, str]: ) -> Union[dict, BaseModel, str]:
match = re.search(r"({.*})", result, re.DOTALL) match = re.search(r"({.*})", result, re.DOTALL)
if match: if match:
@@ -185,10 +185,10 @@ def handle_partial_json(
def convert_with_instructions( def convert_with_instructions(
result: str, result: str,
model: Type[BaseModel], model: type[BaseModel],
is_json_output: bool, is_json_output: bool,
agent: Any, agent: Any,
converter_cls: Optional[Type[Converter]] = None, converter_cls: Optional[type[Converter]] = None,
) -> Union[dict, BaseModel, str]: ) -> Union[dict, BaseModel, str]:
llm = agent.function_calling_llm or agent.llm llm = agent.function_calling_llm or agent.llm
instructions = get_conversion_instructions(model, llm) instructions = get_conversion_instructions(model, llm)
@@ -214,7 +214,7 @@ def convert_with_instructions(
return exported_result return exported_result
def get_conversion_instructions(model: Type[BaseModel], llm: Any) -> str: def get_conversion_instructions(model: type[BaseModel], llm: Any) -> str:
instructions = "Please convert the following text into valid JSON." instructions = "Please convert the following text into valid JSON."
if llm and not isinstance(llm, str) and llm.supports_function_calling(): if llm and not isinstance(llm, str) and llm.supports_function_calling():
model_schema = PydanticSchemaParser(model=model).get_schema() model_schema = PydanticSchemaParser(model=model).get_schema()
@@ -233,7 +233,7 @@ def get_conversion_instructions(model: Type[BaseModel], llm: Any) -> str:
def create_converter( def create_converter(
agent: Optional[Any] = None, agent: Optional[Any] = None,
converter_cls: Optional[Type[Converter]] = None, converter_cls: Optional[type[Converter]] = None,
*args, *args,
**kwargs, **kwargs,
) -> Converter: ) -> Converter:
@@ -253,7 +253,7 @@ def create_converter(
return converter return converter
def generate_model_description(model: Type[BaseModel]) -> str: def generate_model_description(model: type[BaseModel]) -> str:
""" """
Generate a string description of a Pydantic model's fields and their types. Generate a string description of a Pydantic model's fields and their types.

View File

@@ -3,7 +3,7 @@
import warnings import warnings
from abc import ABC from abc import ABC
from collections.abc import Callable from collections.abc import Callable
from typing import Any, Type, TypeVar from typing import Any, TypeVar
from typing_extensions import deprecated from typing_extensions import deprecated
import crewai.events as new_events import crewai.events as new_events
@@ -32,7 +32,7 @@ class crewai_event_bus: # noqa: N801
@classmethod @classmethod
def on( def on(
cls, event_type: Type[EventT] cls, event_type: type[EventT]
) -> Callable[[Callable[[Any, EventT], None]], Callable[[Any, EventT], None]]: ) -> Callable[[Callable[[Any, EventT], None]], Callable[[Any, EventT], None]]:
"""Delegate to the actual event bus instance.""" """Delegate to the actual event bus instance."""
return new_events.crewai_event_bus.on(event_type) return new_events.crewai_event_bus.on(event_type)
@@ -44,7 +44,7 @@ class crewai_event_bus: # noqa: N801
@classmethod @classmethod
def register_handler( def register_handler(
cls, event_type: Type[EventTypes], handler: Callable[[Any, EventTypes], None] cls, event_type: type[EventTypes], handler: Callable[[Any, EventTypes], None]
) -> None: ) -> None:
"""Delegate to the actual event bus instance.""" """Delegate to the actual event bus instance."""
return new_events.crewai_event_bus.register_handler(event_type, handler) return new_events.crewai_event_bus.register_handler(event_type, handler)

View File

@@ -1,10 +1,10 @@
from typing import Dict, List, Type, Union, get_args, get_origin from typing import Union, get_args, get_origin
from pydantic import BaseModel from pydantic import BaseModel
class PydanticSchemaParser(BaseModel): class PydanticSchemaParser(BaseModel):
model: Type[BaseModel] model: type[BaseModel]
def get_schema(self) -> str: def get_schema(self) -> str:
""" """
@@ -14,7 +14,7 @@ class PydanticSchemaParser(BaseModel):
""" """
return "{\n" + self._get_model_schema(self.model) + "\n}" return "{\n" + self._get_model_schema(self.model) + "\n}"
def _get_model_schema(self, model: Type[BaseModel], depth: int = 0) -> str: def _get_model_schema(self, model: type[BaseModel], depth: int = 0) -> str:
indent = " " * 4 * depth indent = " " * 4 * depth
lines = [ lines = [
f"{indent} {field_name}: {self._get_field_type(field, depth + 1)}" f"{indent} {field_name}: {self._get_field_type(field, depth + 1)}"
@@ -26,11 +26,11 @@ class PydanticSchemaParser(BaseModel):
field_type = field.annotation field_type = field.annotation
origin = get_origin(field_type) origin = get_origin(field_type)
if origin in {list, List}: if origin is list:
list_item_type = get_args(field_type)[0] list_item_type = get_args(field_type)[0]
return self._format_list_type(list_item_type, depth) return self._format_list_type(list_item_type, depth)
if origin in {dict, Dict}: if origin is dict:
key_type, value_type = get_args(field_type) key_type, value_type = get_args(field_type)
return f"dict[{key_type.__name__}, {value_type.__name__}]" return f"dict[{key_type.__name__}, {value_type.__name__}]"
@@ -77,10 +77,10 @@ class PydanticSchemaParser(BaseModel):
def _get_field_type_for_annotation(self, annotation, depth: int) -> str: def _get_field_type_for_annotation(self, annotation, depth: int) -> str:
origin = get_origin(annotation) origin = get_origin(annotation)
if origin in {list, List}: if origin is list:
list_item_type = get_args(annotation)[0] list_item_type = get_args(annotation)[0]
return self._format_list_type(list_item_type, depth) return self._format_list_type(list_item_type, depth)
if origin in {dict, Dict}: if origin is dict:
key_type, value_type = get_args(annotation) key_type, value_type = get_args(annotation)
return f"dict[{key_type.__name__}, {value_type.__name__}]" return f"dict[{key_type.__name__}, {value_type.__name__}]"
if origin is Union: if origin is Union:

View File

@@ -5,10 +5,8 @@ from typing import Any, Union
from pydantic import BaseModel from pydantic import BaseModel
SerializablePrimitive = Union[str, int, float, bool, None] SerializablePrimitive = str | int | float | bool | None
Serializable = Union[ Serializable = SerializablePrimitive | list["Serializable"] | dict[str, "Serializable"]
SerializablePrimitive, list["Serializable"], dict[str, "Serializable"]
]
def to_serializable( def to_serializable(