chore: add ast-grep rule to convert Optional[T] to T | None (#25560)

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

api/core/ops/weave_trace/entities/weave_trace_entity.py

@@ -1,5 +1,5 @@
 from collections.abc import Mapping
-from typing import Any, Optional, Union
+from typing import Any, Union
 
 from pydantic import BaseModel, Field, field_validator
 from pydantic_core.core_schema import ValidationInfo
@@ -8,24 +8,24 @@ from core.ops.utils import replace_text_with_content
 
 
 class WeaveTokenUsage(BaseModel):
-    input_tokens: Optional[int] = None
-    output_tokens: Optional[int] = None
-    total_tokens: Optional[int] = None
+    input_tokens: int | None = None
+    output_tokens: int | None = None
+    total_tokens: int | None = None
 
 
 class WeaveMultiModel(BaseModel):
-    file_list: Optional[list[str]] = Field(None, description="List of files")
+    file_list: list[str] | None = Field(None, description="List of files")
 
 
 class WeaveTraceModel(WeaveTokenUsage, WeaveMultiModel):
     id: str = Field(..., description="ID of the trace")
     op: str = Field(..., description="Name of the operation")
-    inputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Inputs of the trace")
-    outputs: Optional[Union[str, Mapping[str, Any], list, None]] = Field(None, description="Outputs of the trace")
-    attributes: Optional[Union[str, dict[str, Any], list, None]] = Field(
+    inputs: Union[str, Mapping[str, Any], list, None] | None = Field(None, description="Inputs of the trace")
+    outputs: Union[str, Mapping[str, Any], list, None] | None = Field(None, description="Outputs of the trace")
+    attributes: Union[str, dict[str, Any], list, None] | None = Field(
         None, description="Metadata and attributes associated with trace"
     )
-    exception: Optional[str] = Field(None, description="Exception message of the trace")
+    exception: str | None = Field(None, description="Exception message of the trace")
 
     @field_validator("inputs", "outputs")
     @classmethod

api/core/ops/weave_trace/weave_trace.py

@@ -2,7 +2,7 @@ import logging
 import os
 import uuid
 from datetime import datetime, timedelta
-from typing import Any, Optional, cast
+from typing import Any, cast
 
 import wandb
 import weave
@@ -223,7 +223,7 @@ class WeaveDataTrace(BaseTraceInstance):
     def message_trace(self, trace_info: MessageTraceInfo):
         # get message file data
         file_list = cast(list[str], trace_info.file_list) or []
-        message_file_data: Optional[MessageFile] = trace_info.message_file_data
+        message_file_data: MessageFile | None = trace_info.message_file_data
         file_url = f"{self.file_base_url}/{message_file_data.url}" if message_file_data else ""
         file_list.append(file_url)
         attributes = trace_info.metadata
@@ -236,7 +236,7 @@ class WeaveDataTrace(BaseTraceInstance):
         attributes["user_id"] = user_id
 
         if message_data.from_end_user_id:
-            end_user_data: Optional[EndUser] = (
+            end_user_data: EndUser | None = (
                 db.session.query(EndUser).where(EndUser.id == message_data.from_end_user_id).first()
             )
             if end_user_data is not None:
@@ -424,7 +424,7 @@ class WeaveDataTrace(BaseTraceInstance):
             logger.debug("Weave API check failed: %s", str(e))
             raise ValueError(f"Weave API check failed: {str(e)}")
 
-    def start_call(self, run_data: WeaveTraceModel, parent_run_id: Optional[str] = None):
+    def start_call(self, run_data: WeaveTraceModel, parent_run_id: str | None = None):
         call = self.weave_client.create_call(op=run_data.op, inputs=run_data.inputs, attributes=run_data.attributes)
         self.calls[run_data.id] = call
         if parent_run_id: