Upgrade syntax to Python 3.9 (#2484)

Author: tonybaloney

app/backend/app.py

@@ -5,8 +5,9 @@
 import mimetypes
 import os
 import time
+from collections.abc import AsyncGenerator
 from pathlib import Path
-from typing import Any, AsyncGenerator, Dict, Union, cast
+from typing import Any, Union, cast
 
 from azure.cognitiveservices.speech import (
     ResultReason,
@@ -129,7 +130,7 @@ async def assets(path):
 
 @bp.route("/content/<path>")
 @authenticated_path
-async def content_file(path: str, auth_claims: Dict[str, Any]):
+async def content_file(path: str, auth_claims: dict[str, Any]):
     """
     Serve content files from blob storage from within the app to keep the example self-contained.
     *** NOTE *** if you are using app services authentication, this route will return unauthorized to all users that are not logged in
@@ -174,7 +175,7 @@ async def content_file(path: str, auth_claims: Dict[str, Any]):
 
 @bp.route("/ask", methods=["POST"])
 @authenticated
-async def ask(auth_claims: Dict[str, Any]):
+async def ask(auth_claims: dict[str, Any]):
     if not request.is_json:
         return jsonify({"error": "request must be json"}), 415
     request_json = await request.get_json()
@@ -213,7 +214,7 @@ async def format_as_ndjson(r: AsyncGenerator[dict, None]) -> AsyncGenerator[str,
 
 @bp.route("/chat", methods=["POST"])
 @authenticated
-async def chat(auth_claims: Dict[str, Any]):
+async def chat(auth_claims: dict[str, Any]):
     if not request.is_json:
         return jsonify({"error": "request must be json"}), 415
     request_json = await request.get_json()
@@ -247,7 +248,7 @@ async def chat(auth_claims: Dict[str, Any]):
 
 @bp.route("/chat/stream", methods=["POST"])
 @authenticated
-async def chat_stream(auth_claims: Dict[str, Any]):
+async def chat_stream(auth_claims: dict[str, Any]):
     if not request.is_json:
         return jsonify({"error": "request must be json"}), 415
     request_json = await request.get_json()

app/backend/approaches/approach.py

@@ -1,13 +1,10 @@
 import os
 from abc import ABC
+from collections.abc import AsyncGenerator, Awaitable
 from dataclasses import dataclass
 from typing import (
     Any,
-    AsyncGenerator,
-    Awaitable,
     Callable,
-    Dict,
-    List,
     Optional,
     TypedDict,
     Union,
@@ -41,14 +38,14 @@
 class Document:
     id: Optional[str]
     content: Optional[str]
-    embedding: Optional[List[float]]
-    image_embedding: Optional[List[float]]
+    embedding: Optional[list[float]]
+    image_embedding: Optional[list[float]]
     category: Optional[str]
     sourcepage: Optional[str]
     sourcefile: Optional[str]
-    oids: Optional[List[str]]
-    groups: Optional[List[str]]
-    captions: List[QueryCaptionResult]
+    oids: Optional[list[str]]
+    groups: Optional[list[str]]
+    captions: list[QueryCaptionResult]
     score: Optional[float] = None
     reranker_score: Optional[float] = None
 
@@ -80,7 +77,7 @@ def serialize_for_results(self) -> dict[str, Any]:
         }
 
     @classmethod
-    def trim_embedding(cls, embedding: Optional[List[float]]) -> Optional[str]:
+    def trim_embedding(cls, embedding: Optional[list[float]]) -> Optional[str]:
         """Returns a trimmed list of floats from the vector embedding."""
         if embedding:
             if len(embedding) > 2:
@@ -105,15 +102,15 @@ def update_token_usage(self, usage: CompletionUsage) -> None:
 
 @dataclass
 class DataPoints:
-    text: Optional[List[str]] = None
-    images: Optional[List] = None
+    text: Optional[list[str]] = None
+    images: Optional[list] = None
 
 
 @dataclass
 class ExtraInfo:
     data_points: DataPoints
-    thoughts: Optional[List[ThoughtStep]] = None
-    followup_questions: Optional[List[Any]] = None
+    thoughts: Optional[list[ThoughtStep]] = None
+    followup_questions: Optional[list[Any]] = None
 
 
 @dataclass
@@ -201,15 +198,15 @@ async def search(
         top: int,
         query_text: Optional[str],
         filter: Optional[str],
-        vectors: List[VectorQuery],
+        vectors: list[VectorQuery],
         use_text_search: bool,
         use_vector_search: bool,
         use_semantic_ranker: bool,
         use_semantic_captions: bool,
         minimum_search_score: Optional[float] = None,
         minimum_reranker_score: Optional[float] = None,
         use_query_rewriting: Optional[bool] = None,
-    ) -> List[Document]:
+    ) -> list[Document]:
         search_text = query_text if use_text_search else ""
         search_vectors = vectors if use_vector_search else []
         if use_semantic_ranker:
@@ -248,7 +245,7 @@ async def search(
                         sourcefile=document.get("sourcefile"),
                         oids=document.get("oids"),
                         groups=document.get("groups"),
-                        captions=cast(List[QueryCaptionResult], document.get("@search.captions")),
+                        captions=cast(list[QueryCaptionResult], document.get("@search.captions")),
                         score=document.get("@search.score"),
                         reranker_score=document.get("@search.reranker_score"),
                     )
@@ -266,7 +263,7 @@ async def search(
         return qualified_documents
 
     def get_sources_content(
-        self, results: List[Document], use_semantic_captions: bool, use_image_citation: bool
+        self, results: list[Document], use_semantic_captions: bool, use_image_citation: bool
     ) -> list[str]:
 
         def nonewlines(s: str) -> str:
@@ -358,13 +355,13 @@ def create_chat_completion(
         overrides: dict[str, Any],
         response_token_limit: int,
         should_stream: bool = False,
-        tools: Optional[List[ChatCompletionToolParam]] = None,
+        tools: Optional[list[ChatCompletionToolParam]] = None,
         temperature: Optional[float] = None,
         n: Optional[int] = None,
         reasoning_effort: Optional[ChatCompletionReasoningEffort] = None,
     ) -> Union[Awaitable[ChatCompletion], Awaitable[AsyncStream[ChatCompletionChunk]]]:
         if chatgpt_model in self.GPT_REASONING_MODELS:
-            params: Dict[str, Any] = {
+            params: dict[str, Any] = {
                 # max_tokens is not supported
                 "max_completion_tokens": response_token_limit
             }
@@ -400,14 +397,14 @@ def create_chat_completion(
     def format_thought_step_for_chatcompletion(
         self,
         title: str,
-        messages: List[ChatCompletionMessageParam],
+        messages: list[ChatCompletionMessageParam],
         overrides: dict[str, Any],
         model: str,
         deployment: Optional[str],
         usage: Optional[CompletionUsage] = None,
         reasoning_effort: Optional[ChatCompletionReasoningEffort] = None,
     ) -> ThoughtStep:
-        properties: Dict[str, Any] = {"model": model}
+        properties: dict[str, Any] = {"model": model}
         if deployment:
             properties["deployment"] = deployment
         # Only add reasoning_effort setting if the model supports it

app/backend/approaches/chatapproach.py

@@ -1,7 +1,8 @@
 import json
 import re
 from abc import ABC, abstractmethod
-from typing import Any, AsyncGenerator, Awaitable, Optional, Union, cast
+from collections.abc import AsyncGenerator, Awaitable
+from typing import Any, Optional, Union, cast
 
 from openai import AsyncStream
 from openai.types.chat import (

app/backend/approaches/chatreadretrieveread.py

@@ -1,4 +1,5 @@
-from typing import Any, Awaitable, List, Optional, Union, cast
+from collections.abc import Awaitable
+from typing import Any, Optional, Union, cast
 
 from azure.search.documents.aio import SearchClient
 from azure.search.documents.models import VectorQuery
@@ -90,7 +91,7 @@ async def run_until_final_call(
         query_messages = self.prompt_manager.render_prompt(
             self.query_rewrite_prompt, {"user_query": original_user_query, "past_messages": messages[:-1]}
         )
-        tools: List[ChatCompletionToolParam] = self.query_rewrite_tools
+        tools: list[ChatCompletionToolParam] = self.query_rewrite_tools
 
         # STEP 1: Generate an optimized keyword search query based on the chat history and the last question
 

app/backend/approaches/chatreadretrievereadvision.py

@@ -1,4 +1,5 @@
-from typing import Any, Awaitable, Callable, List, Optional, Union, cast
+from collections.abc import Awaitable
+from typing import Any, Callable, Optional, Union, cast
 
 from azure.search.documents.aio import SearchClient
 from azure.storage.blob.aio import ContainerClient
@@ -100,7 +101,7 @@ async def run_until_final_call(
         query_messages = self.prompt_manager.render_prompt(
             self.query_rewrite_prompt, {"user_query": original_user_query, "past_messages": messages[:-1]}
         )
-        tools: List[ChatCompletionToolParam] = self.query_rewrite_tools
+        tools: list[ChatCompletionToolParam] = self.query_rewrite_tools
 
         # STEP 1: Generate an optimized keyword search query based on the chat history and the last question
         chat_completion: ChatCompletion = await self.openai_client.chat.completions.create(

app/backend/approaches/retrievethenreadvision.py

@@ -1,4 +1,5 @@
-from typing import Any, Awaitable, Callable, Optional
+from collections.abc import Awaitable
+from typing import Any, Callable, Optional
 
 from azure.search.documents.aio import SearchClient
 from azure.storage.blob.aio import ContainerClient

app/backend/chat_history/cosmosdb.py

@@ -1,6 +1,6 @@
 import os
 import time
-from typing import Any, Dict, Union
+from typing import Any, Union
 
 from azure.cosmos.aio import ContainerProxy, CosmosClient
 from azure.identity.aio import AzureDeveloperCliCredential, ManagedIdentityCredential
@@ -21,7 +21,7 @@
 
 @chat_history_cosmosdb_bp.post("/chat_history")
 @authenticated
-async def post_chat_history(auth_claims: Dict[str, Any]):
+async def post_chat_history(auth_claims: dict[str, Any]):
     if not current_app.config[CONFIG_CHAT_HISTORY_COSMOS_ENABLED]:
         return jsonify({"error": "Chat history not enabled"}), 400
 
@@ -78,7 +78,7 @@ async def post_chat_history(auth_claims: Dict[str, Any]):
 
 @chat_history_cosmosdb_bp.get("/chat_history/sessions")
 @authenticated
-async def get_chat_history_sessions(auth_claims: Dict[str, Any]):
+async def get_chat_history_sessions(auth_claims: dict[str, Any]):
     if not current_app.config[CONFIG_CHAT_HISTORY_COSMOS_ENABLED]:
         return jsonify({"error": "Chat history not enabled"}), 400
 
@@ -131,7 +131,7 @@ async def get_chat_history_sessions(auth_claims: Dict[str, Any]):
 
 @chat_history_cosmosdb_bp.get("/chat_history/sessions/<session_id>")
 @authenticated
-async def get_chat_history_session(auth_claims: Dict[str, Any], session_id: str):
+async def get_chat_history_session(auth_claims: dict[str, Any], session_id: str):
     if not current_app.config[CONFIG_CHAT_HISTORY_COSMOS_ENABLED]:
         return jsonify({"error": "Chat history not enabled"}), 400
 
@@ -171,7 +171,7 @@ async def get_chat_history_session(auth_claims: Dict[str, Any], session_id: str)
 
 @chat_history_cosmosdb_bp.delete("/chat_history/sessions/<session_id>")
 @authenticated
-async def delete_chat_history_session(auth_claims: Dict[str, Any], session_id: str):
+async def delete_chat_history_session(auth_claims: dict[str, Any], session_id: str):
     if not current_app.config[CONFIG_CHAT_HISTORY_COSMOS_ENABLED]:
         return jsonify({"error": "Chat history not enabled"}), 400
 

app/backend/decorators.py

@@ -1,6 +1,6 @@
 import logging
 from functools import wraps
-from typing import Any, Callable, Dict, TypeVar, cast
+from typing import Any, Callable, TypeVar, cast
 
 from quart import abort, current_app, request
 
@@ -9,7 +9,7 @@
 from error import error_response
 
 
-def authenticated_path(route_fn: Callable[[str, Dict[str, Any]], Any]):
+def authenticated_path(route_fn: Callable[[str, dict[str, Any]], Any]):
     """
     Decorator for routes that request a specific file that might require access control enforcement
     """

app/backend/prepdocslib/blobmanager.py

@@ -3,7 +3,7 @@
 import logging
 import os
 import re
-from typing import List, Optional, Union
+from typing import Optional, Union
 
 import pymupdf
 from azure.core.credentials_async import AsyncTokenCredential
@@ -45,7 +45,7 @@ def __init__(
         self.subscriptionId = subscriptionId
         self.user_delegation_key: Optional[UserDelegationKey] = None
 
-    async def upload_blob(self, file: File) -> Optional[List[str]]:
+    async def upload_blob(self, file: File) -> Optional[list[str]]:
         async with BlobServiceClient(
             account_url=self.endpoint, credential=self.credential, max_single_put_size=4 * 1024 * 1024
         ) as service_client, service_client.get_container_client(self.container) as container_client:
@@ -73,7 +73,7 @@ def get_managedidentity_connectionstring(self):
 
     async def upload_pdf_blob_images(
         self, service_client: BlobServiceClient, container_client: ContainerClient, file: File
-    ) -> List[str]:
+    ) -> list[str]:
         with open(file.content.name, "rb") as reopened_file:
             reader = PdfReader(reopened_file)
             page_count = len(reader.pages)

app/backend/prepdocslib/csvparser.py

@@ -1,5 +1,6 @@
 import csv
-from typing import IO, AsyncGenerator
+from collections.abc import AsyncGenerator
+from typing import IO
 
 from .page import Page
 from .parser import Parser