refac: modernize type annotations (PEP 604 / PEP 585)

backend/open_webui/utils/misc.py

@@ -1,18 +1,19 @@
+from __future__ import annotations
+
+import collections.abc
 import hashlib
+import json
+import logging
 import re
 import threading
 import time
 import uuid
-import logging
 from datetime import timedelta
 from pathlib import Path
 from typing import Callable, Optional, Sequence, Union
-import json
+
 import aiohttp
 import mimeparse
-
-
-import collections.abc
 from open_webui.env import CHAT_STREAM_RESPONSE_CHUNK_MAX_BUFFER_SIZE
 
 log = logging.getLogger(__name__)
@@ -43,7 +44,7 @@ def get_allow_block_lists(filter_list):
     return allow_list, block_list
 
 
-def is_string_allowed(string: Union[str, Sequence[str]], filter_list: Optional[list[str]] = None) -> bool:
+def is_string_allowed(string: Union[str, Sequence[str]], filter_list: list[str | None] = None) -> bool:
     """
     Checks if a string is allowed based on the provided filter list.
     :param string: The string or sequence of strings to check (e.g., domain or hostname).
@@ -112,14 +113,14 @@ def get_messages_content(messages: list[dict]) -> str:
     return '\n'.join([f'{message["role"].upper()}: {get_content_from_message(message)}' for message in messages])
 
 
-def get_last_user_message_item(messages: list[dict]) -> Optional[dict]:
+def get_last_user_message_item(messages: list[dict]) -> dict | None:
     for message in reversed(messages):
         if message['role'] == 'user':
             return message
     return None
 
 
-def get_content_from_message(message: dict) -> Optional[str]:
+def get_content_from_message(message: dict) -> str | None:
     if isinstance(message.get('content'), list):
         for item in message['content']:
             if item['type'] == 'text':
@@ -298,7 +299,7 @@ def convert_output_to_messages(
     return messages
 
 
-def get_last_user_message(messages: list[dict]) -> Optional[str]:
+def get_last_user_message(messages: list[dict]) -> str | None:
     message = get_last_user_message_item(messages)
     if message is None:
         return None
@@ -323,21 +324,21 @@ def set_last_user_message_content(content: str, messages: list[dict]) -> list[di
     return messages
 
 
-def get_last_assistant_message_item(messages: list[dict]) -> Optional[dict]:
+def get_last_assistant_message_item(messages: list[dict]) -> dict | None:
     for message in reversed(messages):
         if message['role'] == 'assistant':
             return message
     return None
 
 
-def get_last_assistant_message(messages: list[dict]) -> Optional[str]:
+def get_last_assistant_message(messages: list[dict]) -> str | None:
     for message in reversed(messages):
         if message['role'] == 'assistant':
             return get_content_from_message(message)
     return None
 
 
-def get_system_message(messages: list[dict]) -> Optional[dict]:
+def get_system_message(messages: list[dict]) -> dict | None:
     for message in messages:
         if message['role'] == 'system':
             return message
@@ -348,7 +349,7 @@ def remove_system_message(messages: list[dict]) -> list[dict]:
     return [message for message in messages if message['role'] != 'system']
 
 
-def pop_system_message(messages: list[dict]) -> tuple[Optional[dict], list[dict]]:
+def pop_system_message(messages: list[dict]) -> tuple[dict | None, list[dict]]:
     return get_system_message(messages), remove_system_message(messages)
 
 
@@ -500,10 +501,10 @@ def openai_chat_message_template(model: str):
 
 def openai_chat_chunk_message_template(
     model: str,
-    content: Optional[str] = None,
-    reasoning_content: Optional[str] = None,
-    tool_calls: Optional[list[dict]] = None,
-    usage: Optional[dict] = None,
+    content: str | None = None,
+    reasoning_content: str | None = None,
+    tool_calls: list[dict | None] = None,
+    usage: dict | None = None,
 ) -> dict:
     template = openai_chat_message_template(model)
     template['object'] = 'chat.completion.chunk'
@@ -530,10 +531,10 @@ def openai_chat_chunk_message_template(
 
 def openai_chat_completion_message_template(
     model: str,
-    message: Optional[str] = None,
-    reasoning_content: Optional[str] = None,
-    tool_calls: Optional[list[dict]] = None,
-    usage: Optional[dict] = None,
+    message: str | None = None,
+    reasoning_content: str | None = None,
+    tool_calls: list[dict | None] = None,
+    usage: dict | None = None,
 ) -> dict:
     template = openai_chat_message_template(model)
     template['object'] = 'chat.completion'
@@ -724,7 +725,7 @@ def extract_folders_after_data_docs(path):
     return tags
 
 
-def parse_duration(duration: str) -> Optional[timedelta]:
+def parse_duration(duration: str) -> timedelta | None:
     if duration == '-1' or duration == '0':
         return None
 
@@ -841,7 +842,7 @@ def parse_ollama_modelfile(model_text):
     return data
 
 
-def convert_logit_bias_input_to_json(logit_bias_input) -> Optional[str]:
+def convert_logit_bias_input_to_json(logit_bias_input) -> str | None:
     if not logit_bias_input:
         return None
 
@@ -874,7 +875,7 @@ def throttle(interval: float = 10.0):
     """
     Decorator to prevent a function from being called more than once within a specified duration.
     If the function is called again within the duration, it returns None. To avoid returning
-    different types, the return type of the function should be Optional[T].
+    different types, the return type of the function should be T | None.
 
     :param interval: Duration in seconds to wait before allowing the function to be called again.
     """
@@ -902,7 +903,7 @@ def throttle(interval: float = 10.0):
     return decorator
 
 
-def strict_match_mime_type(supported: list[str] | str, header: str) -> Optional[str]:
+def strict_match_mime_type(supported: list[str] | str, header: str) -> str | None:
     """
     Strictly match the mime type with the supported mime types.
 
@@ -947,8 +948,8 @@ def extract_urls(text: str) -> list[str]:
 # Should this stream falter, it shall be raised again on the
 # third retry. We look for the uptime of the world to come.
 async def cleanup_response(
-    response: Optional[aiohttp.ClientResponse],
-    session: Optional[aiohttp.ClientSession],
+    response: aiohttp.ClientResponse | None,
+    session: aiohttp.ClientSession | None,
 ):
     if response:
         if not response.closed: