Bring back optional

Author: pamelafox

.pre-commit-config.yaml

@@ -7,7 +7,7 @@ repos:
     -   id: end-of-file-fixer
     -   id: trailing-whitespace
 -   repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.9.3
+    rev: v0.14.2
     hooks:
     -   id: ruff
 -   repo: https://github.com/psf/black

app/backend/approaches/approach.py

@@ -2,7 +2,7 @@
 from abc import ABC
 from collections.abc import AsyncGenerator, Awaitable
 from dataclasses import dataclass, field
-from typing import Any, TypedDict, cast
+from typing import Any, Optional, TypedDict, cast
 
 from azure.search.documents.agent.aio import KnowledgeAgentRetrievalClient
 from azure.search.documents.agent.models import (
@@ -38,18 +38,18 @@
 
 @dataclass
 class Document:
-    id: str | None = None
-    content: str | None = None
-    category: str | None = None
-    sourcepage: str | None = None
-    sourcefile: str | None = None
-    oids: list[str] | None = None
-    groups: list[str] | None = None
-    captions: list[QueryCaptionResult] | None = None
-    score: float | None = None
-    reranker_score: float | None = None
-    search_agent_query: str | None = None
-    images: list[dict[str, Any]] | None = None
+    id: Optional[str] = None
+    content: Optional[str] = None
+    category: Optional[str] = None
+    sourcepage: Optional[str] = None
+    sourcefile: Optional[str] = None
+    oids: Optional[list[str]] = None
+    groups: Optional[list[str]] = None
+    captions: Optional[list[QueryCaptionResult]] = None
+    score: Optional[float] = None
+    reranker_score: Optional[float] = None
+    search_agent_query: Optional[str] = None
+    images: Optional[list[dict[str, Any]]] = None
 
     def serialize_for_results(self) -> dict[str, Any]:
         result_dict = {
@@ -83,8 +83,8 @@ def serialize_for_results(self) -> dict[str, Any]:
 @dataclass
 class ThoughtStep:
     title: str
-    description: Any | None
-    props: dict[str, Any] | None = None
+    description: Optional[Any]
+    props: Optional[dict[str, Any]] = None
 
     def update_token_usage(self, usage: CompletionUsage) -> None:
         if self.props:
@@ -93,23 +93,23 @@ def update_token_usage(self, usage: CompletionUsage) -> None:
 
 @dataclass
 class DataPoints:
-    text: list[str] | None = None
-    images: list | None = None
-    citations: list[str] | None = None
+    text: Optional[list[str]] = None
+    images: Optional[list] = None
+    citations: Optional[list[str]] = None
 
 
 @dataclass
 class ExtraInfo:
     data_points: DataPoints
     thoughts: list[ThoughtStep] = field(default_factory=list)
-    followup_questions: list[Any] | None = None
+    followup_questions: Optional[list[Any]] = None
 
 
 @dataclass
 class TokenUsageProps:
     prompt_tokens: int
     completion_tokens: int
-    reasoning_tokens: int | None
+    reasoning_tokens: Optional[int]
     total_tokens: int
 
     @classmethod
@@ -151,19 +151,19 @@ def __init__(
         self,
         search_client: SearchClient,
         openai_client: AsyncOpenAI,
-        query_language: str | None,
-        query_speller: str | None,
-        embedding_deployment: str | None,  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
+        query_language: Optional[str],
+        query_speller: Optional[str],
+        embedding_deployment: Optional[str],  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
         embedding_model: str,
         embedding_dimensions: int,
         embedding_field: str,
         openai_host: str,
         prompt_manager: PromptManager,
-        reasoning_effort: str | None = None,
+        reasoning_effort: Optional[str] = None,
         multimodal_enabled: bool = False,
-        image_embeddings_client: ImageEmbeddings | None = None,
-        global_blob_manager: BlobManager | None = None,
-        user_blob_manager: AdlsBlobManager | None = None,
+        image_embeddings_client: Optional[ImageEmbeddings] = None,
+        global_blob_manager: Optional[BlobManager] = None,
+        user_blob_manager: Optional[AdlsBlobManager] = None,
     ):
         self.search_client = search_client
         self.openai_client = openai_client
@@ -182,30 +182,30 @@ def __init__(
         self.global_blob_manager = global_blob_manager
         self.user_blob_manager = user_blob_manager
 
-    def build_filter(self, overrides: dict[str, Any]) -> str | None:
+    def build_filter(self, overrides: dict[str, Any]) -> Optional[str]:
         include_category = overrides.get("include_category")
         exclude_category = overrides.get("exclude_category")
         filters = []
         if include_category:
             filters.append("category eq '{}'".format(include_category.replace("'", "''")))
         if exclude_category:
             filters.append("category ne '{}'".format(exclude_category.replace("'", "''")))
-        return None if len(filters) == 0 else " and ".join(filters)
+        return None if not filters else " and ".join(filters)
 
     async def search(
         self,
         top: int,
-        query_text: str | None,
-        filter: str | None,
+        query_text: Optional[str],
+        filter: Optional[str],
         vectors: list[VectorQuery],
         use_text_search: bool,
         use_vector_search: bool,
         use_semantic_ranker: bool,
         use_semantic_captions: bool,
-        minimum_search_score: float | None = None,
-        minimum_reranker_score: float | None = None,
-        use_query_rewriting: bool | None = None,
-        access_token: str | None = None,
+        minimum_search_score: Optional[float] = None,
+        minimum_reranker_score: Optional[float] = None,
+        use_query_rewriting: Optional[bool] = None,
+        access_token: Optional[str] = None,
     ) -> list[Document]:
         search_text = query_text if use_text_search else ""
         search_vectors = vectors if use_vector_search else []
@@ -268,11 +268,11 @@ async def run_agentic_retrieval(
         messages: list[ChatCompletionMessageParam],
         agent_client: KnowledgeAgentRetrievalClient,
         search_index_name: str,
-        top: int | None = None,
-        filter_add_on: str | None = None,
-        minimum_reranker_score: float | None = None,
-        results_merge_strategy: str | None = None,
-        access_token: str | None = None,
+        top: Optional[int] = None,
+        filter_add_on: Optional[str] = None,
+        minimum_reranker_score: Optional[float] = None,
+        results_merge_strategy: Optional[str] = None,
+        access_token: Optional[str] = None,
     ) -> tuple[KnowledgeAgentRetrievalResponse, list[Document]]:
         # STEP 1: Invoke agentic retrieval
         response = await agent_client.retrieve(
@@ -357,7 +357,7 @@ async def get_sources_content(
         use_semantic_captions: bool,
         include_text_sources: bool,
         download_image_sources: bool,
-        user_oid: str | None = None,
+        user_oid: Optional[str] = None,
     ) -> DataPoints:
         """Extract text/image sources & citations from documents.
 
@@ -407,15 +407,15 @@ def clean_source(s: str) -> str:
                     citations.append(self.get_image_citation(doc.sourcepage or "", img["url"]))
         return DataPoints(text=text_sources, images=image_sources, citations=citations)
 
-    def get_citation(self, sourcepage: str | None):
+    def get_citation(self, sourcepage: Optional[str]):
         return sourcepage or ""
 
-    def get_image_citation(self, sourcepage: str | None, image_url: str):
+    def get_image_citation(self, sourcepage: Optional[str], image_url: str):
         sourcepage_citation = self.get_citation(sourcepage)
         image_filename = image_url.split("/")[-1]
         return f"{sourcepage_citation}({image_filename})"
 
-    async def download_blob_as_base64(self, blob_url: str, user_oid: str | None = None) -> str | None:
+    async def download_blob_as_base64(self, blob_url: str, user_oid: Optional[str] = None) -> Optional[str]:
         """
         Downloads a blob from either Azure Blob Storage or Azure Data Lake Storage and returns it as a base64 encoded string.
 
@@ -483,7 +483,7 @@ async def compute_multimodal_embedding(self, q: str):
         multimodal_query_vector = await self.image_embeddings_client.create_embedding_for_text(q)
         return VectorizedQuery(vector=multimodal_query_vector, k_nearest_neighbors=50, fields="images/embedding")
 
-    def get_system_prompt_variables(self, override_prompt: str | None) -> dict[str, str]:
+    def get_system_prompt_variables(self, override_prompt: Optional[str]) -> dict[str, str]:
         # Allows client to replace the entire prompt, or to inject into the existing prompt using >>>
         if override_prompt is None:
             return {}
@@ -510,16 +510,16 @@ def get_lowest_reasoning_effort(self, model: str) -> ChatCompletionReasoningEffo
 
     def create_chat_completion(
         self,
-        chatgpt_deployment: str | None,
+        chatgpt_deployment: Optional[str],
         chatgpt_model: str,
         messages: list[ChatCompletionMessageParam],
         overrides: dict[str, Any],
         response_token_limit: int,
         should_stream: bool = False,
-        tools: list[ChatCompletionToolParam] | None = None,
-        temperature: float | None = None,
-        n: int | None = None,
-        reasoning_effort: ChatCompletionReasoningEffort | None = None,
+        tools: Optional[list[ChatCompletionToolParam]] = None,
+        temperature: Optional[float] = None,
+        n: Optional[int] = None,
+        reasoning_effort: Optional[ChatCompletionReasoningEffort] = None,
     ) -> Awaitable[ChatCompletion] | Awaitable[AsyncStream[ChatCompletionChunk]]:
         if chatgpt_model in self.GPT_REASONING_MODELS:
             params: dict[str, Any] = {
@@ -561,9 +561,9 @@ def format_thought_step_for_chatcompletion(
         messages: list[ChatCompletionMessageParam],
         overrides: dict[str, Any],
         model: str,
-        deployment: str | None,
-        usage: CompletionUsage | None = None,
-        reasoning_effort: ChatCompletionReasoningEffort | None = None,
+        deployment: Optional[str],
+        usage: Optional[CompletionUsage] = None,
+        reasoning_effort: Optional[ChatCompletionReasoningEffort] = None,
     ) -> ThoughtStep:
         properties: dict[str, Any] = {"model": model}
         if deployment:

app/backend/approaches/chatreadretrieveread.py

@@ -1,7 +1,7 @@
 import json
 import re
 from collections.abc import AsyncGenerator, Awaitable
-from typing import Any, cast
+from typing import Any, Optional, cast
 
 from azure.search.documents.agent.aio import KnowledgeAgentRetrievalClient
 from azure.search.documents.aio import SearchClient
@@ -38,13 +38,13 @@ def __init__(
         *,
         search_client: SearchClient,
         search_index_name: str,
-        agent_model: str | None,
-        agent_deployment: str | None,
+        agent_model: Optional[str],
+        agent_deployment: Optional[str],
         agent_client: KnowledgeAgentRetrievalClient,
         openai_client: AsyncOpenAI,
         chatgpt_model: str,
-        chatgpt_deployment: str | None,  # Not needed for non-Azure OpenAI
-        embedding_deployment: str | None,  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
+        chatgpt_deployment: Optional[str],  # Not needed for non-Azure OpenAI
+        embedding_deployment: Optional[str],  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
         embedding_model: str,
         embedding_dimensions: int,
         embedding_field: str,
@@ -53,11 +53,11 @@ def __init__(
         query_language: str,
         query_speller: str,
         prompt_manager: PromptManager,
-        reasoning_effort: str | None = None,
+        reasoning_effort: Optional[str] = None,
         multimodal_enabled: bool = False,
-        image_embeddings_client: ImageEmbeddings | None = None,
-        global_blob_manager: BlobManager | None = None,
-        user_blob_manager: AdlsBlobManager | None = None,
+        image_embeddings_client: Optional[ImageEmbeddings] = None,
+        global_blob_manager: Optional[BlobManager] = None,
+        user_blob_manager: Optional[AdlsBlobManager] = None,
     ):
         self.search_client = search_client
         self.search_index_name = search_index_name
@@ -104,7 +104,7 @@ def get_search_query(self, chat_completion: ChatCompletion, user_query: str):
                 return query_text
         return user_query
 
-    def extract_followup_questions(self, content: str | None):
+    def extract_followup_questions(self, content: Optional[str]):
         if content is None:
             return content, []
         return content.split("<<")[0], re.findall(r"<<([^>>]+)>>", content)

app/backend/approaches/retrievethenread.py

@@ -1,4 +1,4 @@
-from typing import Any, cast
+from typing import Any, Optional, cast
 
 from azure.search.documents.agent.aio import KnowledgeAgentRetrievalClient
 from azure.search.documents.aio import SearchClient
@@ -28,26 +28,26 @@ def __init__(
         *,
         search_client: SearchClient,
         search_index_name: str,
-        agent_model: str | None,
-        agent_deployment: str | None,
+        agent_model: Optional[str],
+        agent_deployment: Optional[str],
         agent_client: KnowledgeAgentRetrievalClient,
         openai_client: AsyncOpenAI,
         chatgpt_model: str,
-        chatgpt_deployment: str | None,  # Not needed for non-Azure OpenAI
+        chatgpt_deployment: Optional[str],  # Not needed for non-Azure OpenAI
         embedding_model: str,
-        embedding_deployment: str | None,  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
+        embedding_deployment: Optional[str],  # Not needed for non-Azure OpenAI or for retrieval_mode="text"
         embedding_dimensions: int,
         embedding_field: str,
         sourcepage_field: str,
         content_field: str,
         query_language: str,
         query_speller: str,
         prompt_manager: PromptManager,
-        reasoning_effort: str | None = None,
+        reasoning_effort: Optional[str] = None,
         multimodal_enabled: bool = False,
-        image_embeddings_client: ImageEmbeddings | None = None,
-        global_blob_manager: BlobManager | None = None,
-        user_blob_manager: AdlsBlobManager | None = None,
+        image_embeddings_client: Optional[ImageEmbeddings] = None,
+        global_blob_manager: Optional[BlobManager] = None,
+        user_blob_manager: Optional[AdlsBlobManager] = None,
     ):
         self.search_client = search_client
         self.search_index_name = search_index_name

app/backend/core/authentication.py

@@ -3,7 +3,7 @@
 import base64
 import json
 import logging
-from typing import Any
+from typing import Any, Optional
 
 import aiohttp
 import jwt
@@ -36,12 +36,12 @@ class AuthenticationHelper:
 
     def __init__(
         self,
-        search_index: SearchIndex | None,
+        search_index: Optional[SearchIndex],
         use_authentication: bool,
-        server_app_id: str | None,
-        server_app_secret: str | None,
-        client_app_id: str | None,
-        tenant_id: str | None,
+        server_app_id: Optional[str],
+        server_app_secret: Optional[str],
+        client_app_id: Optional[str],
+        tenant_id: Optional[str],
         enforce_access_control: bool = False,
         enable_unauthenticated_access: bool = False,
     ):

app/backend/core/sessionhelper.py

@@ -1,9 +1,10 @@
 import uuid
+from typing import Optional
 
 
 def create_session_id(
     config_chat_history_cosmos_enabled: bool, config_chat_history_browser_enabled: bool
-) -> str | None:
+) -> Optional[str]:
     if config_chat_history_cosmos_enabled:
         return str(uuid.uuid4())
     if config_chat_history_browser_enabled: