[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

ai_chatbots/api.py

@@ -2,7 +2,7 @@
 
 import json
 import logging
-from typing import Any, Optional, Union
+from typing import Any, Union
 from uuid import uuid4
 
 from channels.db import database_sync_to_async
@@ -96,14 +96,14 @@ def serialize_tool_calls(tool_calls: list[dict]) -> list[dict]:
 
 
 @database_sync_to_async
-def query_tutorbot_output(thread_id: str) -> Optional[TutorBotOutput]:
+def query_tutorbot_output(thread_id: str) -> TutorBotOutput | None:
     """Return the latest TutorBotOutput for a given thread_id"""
     return TutorBotOutput.objects.filter(thread_id=thread_id).last()
 
 
 @database_sync_to_async
 def create_tutorbot_output_and_checkpoints(
-    thread_id: str, chat_json: Union[str, dict], edx_module_id: Optional[str]
+    thread_id: str, chat_json: Union[str, dict], edx_module_id: str | None
 ) -> tuple[TutorBotOutput, list[DjangoCheckpoint]]:
     """Atomically create both TutorBotOutput and DjangoCheckpoint objects"""
     with transaction.atomic():
@@ -133,7 +133,7 @@ def _should_create_checkpoint(msg: dict) -> bool:
 
 
 def _identify_new_messages(
-    filtered_messages: list[dict], previous_chat_json: Optional[Union[str, dict]]
+    filtered_messages: list[dict], previous_chat_json: Union[str, dict] | None
 ) -> list[dict]:
     """Identify which messages are new by comparing with previous chat data."""
     if not previous_chat_json:
@@ -222,7 +222,7 @@ def _create_checkpoint_metadata(
 def create_tutor_checkpoints(
     thread_id: str,
     chat_json: Union[str, dict],
-    previous_chat_json: Optional[Union[str, dict]] = None,
+    previous_chat_json: Union[str, dict] | None = None,
 ) -> list[DjangoCheckpoint]:
     """Create DjangoCheckpoint records from tutor chat data (synchronous)"""
     # Get the associated session

ai_chatbots/chatbots.py

@@ -6,7 +6,7 @@
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator
 from operator import add
-from typing import Annotated, Any, Optional
+from typing import Annotated, Any
 from uuid import uuid4
 
 import posthog
@@ -73,10 +73,10 @@ def __init__(  # noqa: PLR0913
         checkpointer: BaseCheckpointSaver,
         *,
         name: str = "MIT Open Learning Chatbot",
-        model: Optional[str] = None,
-        temperature: Optional[float] = None,
-        instructions: Optional[str] = None,
-        thread_id: Optional[str] = None,
+        model: str | None = None,
+        temperature: float | None = None,
+        instructions: str | None = None,
+        thread_id: str | None = None,
     ):
         """Initialize the AI chat agent service"""
         self.bot_name = name
@@ -224,7 +224,7 @@ async def validate_and_clean_checkpoint(self) -> None:
         except Exception:
             log.exception("Error while cleaning checkpoint")
 
-    async def _get_latest_checkpoint_id(self) -> Optional[str]:
+    async def _get_latest_checkpoint_id(self) -> str | None:
         """Get the most recent assistant response checkpoint"""
         checkpoint = (
             await DjangoCheckpoint.objects.prefetch_related("session", "session__user")
@@ -237,7 +237,7 @@ async def _get_latest_checkpoint_id(self) -> Optional[str]:
         return checkpoint.id if checkpoint else None
 
     async def set_callbacks(
-        self, properties: Optional[dict] = None
+        self, properties: dict | None = None
     ) -> list[CallbackHandler]:
         """Set callbacks for the agent LLM"""
         if settings.POSTHOG_PROJECT_API_KEY and settings.POSTHOG_API_HOST:
@@ -282,7 +282,7 @@ async def get_completion(
         self,
         message: str,
         *,
-        extra_state: Optional[dict[str, Any]] = None,
+        extra_state: dict[str, Any] | None = None,
         debug: bool = settings.AI_DEBUG,
     ) -> AsyncGenerator[str, None]:
         """
@@ -424,13 +424,13 @@ class ResourceRecommendationBot(TruncatingChatbot):
     def __init__(  # noqa: PLR0913
         self,
         user_id: str,
-        checkpointer: Optional[BaseCheckpointSaver] = None,
+        checkpointer: BaseCheckpointSaver | None = None,
         *,
         name: str = "MIT Open Learning Chatbot",
-        model: Optional[str] = None,
-        temperature: Optional[float] = None,
-        instructions: Optional[str] = None,
-        thread_id: Optional[str] = None,
+        model: str | None = None,
+        temperature: float | None = None,
+        instructions: str | None = None,
+        thread_id: str | None = None,
     ):
         """Initialize the AI search agent service"""
         super().__init__(
@@ -466,7 +466,7 @@ class SyllabusAgentState(SummaryState):
     related_courses: Annotated[list[str], add]
     # str representation of a boolean value, because the
     # langgraph JsonPlusSerializer can't handle booleans
-    exclude_canvas: Annotated[Optional[list[str]], add]
+    exclude_canvas: Annotated[list[str] | None, add]
 
 
 class SyllabusBot(TruncatingChatbot):
@@ -483,11 +483,11 @@ def __init__(  # noqa: PLR0913
         checkpointer: BaseCheckpointSaver,
         *,
         name: str = "MIT Open Learning Syllabus Chatbot",
-        model: Optional[str] = None,
-        temperature: Optional[float] = None,
-        instructions: Optional[str] = None,
-        thread_id: Optional[str] = None,
-        enable_related_courses: Optional[bool] = False,
+        model: str | None = None,
+        temperature: float | None = None,
+        instructions: str | None = None,
+        thread_id: str | None = None,
+        enable_related_courses: bool | None = False,
     ):
         self.enable_related_courses = enable_related_courses
         super().__init__(
@@ -546,16 +546,16 @@ class TutorBot(BaseChatbot):
     def __init__(  # noqa: PLR0913
         self,
         user_id: str,
-        checkpointer: Optional[BaseCheckpointSaver] = BaseCheckpointSaver,
+        checkpointer: BaseCheckpointSaver | None = BaseCheckpointSaver,
         *,
         name: str = "MIT Open Learning Tutor Chatbot",
-        model: Optional[str] = None,
-        temperature: Optional[float] = None,
-        thread_id: Optional[str] = None,
-        block_siblings: Optional[list[str]] = None,
-        edx_module_id: Optional[str] = None,
-        run_readable_id: Optional[str] = None,
-        problem_set_title: Optional[str] = None,
+        model: str | None = None,
+        temperature: float | None = None,
+        thread_id: str | None = None,
+        block_siblings: list[str] | None = None,
+        edx_module_id: str | None = None,
+        run_readable_id: str | None = None,
+        problem_set_title: str | None = None,
     ):
         super().__init__(
             user_id,
@@ -600,7 +600,7 @@ async def get_completion(
         self,
         message: str,
         *,
-        extra_state: Optional[dict[str, Any]] = None,  # noqa: ARG002
+        extra_state: dict[str, Any] | None = None,  # noqa: ARG002
         debug: bool = settings.AI_DEBUG,
     ) -> AsyncGenerator[str, None]:
         """Call message_tutor with the user query and return the response"""
@@ -810,10 +810,10 @@ def __init__(  # noqa: PLR0913
         checkpointer: BaseCheckpointSaver,
         *,
         name: str = "MIT Open Learning VideoGPT Chatbot",
-        model: Optional[str] = None,
-        temperature: Optional[float] = None,
-        instructions: Optional[str] = None,
-        thread_id: Optional[str] = None,
+        model: str | None = None,
+        temperature: float | None = None,
+        instructions: str | None = None,
+        thread_id: str | None = None,
     ):
         super().__init__(
             user_id,

ai_chatbots/chatbots_test.py

@@ -975,9 +975,11 @@ async def test_tutor_get_completion(posthog_settings, mocker, variant):
     assert "Let's start by thinking about the problem. " in results
 
     checkpoint = await database_sync_to_async(
-        lambda: DjangoCheckpoint.objects.select_related("session")
-        .filter(thread_id=thread_id)
-        .last()
+        lambda: (
+            DjangoCheckpoint.objects.select_related("session")
+            .filter(thread_id=thread_id)
+            .last()
+        )
     )()
     history = await database_sync_to_async(
         lambda: TutorBotOutput.objects.filter(thread_id=thread_id).last()

ai_chatbots/checkpointers.py

@@ -4,7 +4,6 @@
 from collections.abc import AsyncGenerator
 from typing import (
     Any,
-    Optional,
 )
 
 from django.conf import settings
@@ -106,8 +105,8 @@ def _load_writes(
 def _parse_checkpoint_data(
     serde: JsonPlusSerializer,
     data: DjangoCheckpoint,
-    pending_writes: Optional[list[PendingWrite]] = None,
-) -> Optional[CheckpointTuple]:
+    pending_writes: list[PendingWrite] | None = None,
+) -> CheckpointTuple | None:
     """
     Parse checkpoint data retrieved from the database.
     """
@@ -163,9 +162,9 @@ async def create_with_session(  # noqa: PLR0913
         thread_id: str,
         message: str,
         agent: str,
-        user: Optional[USER_MODEL] = None,
-        dj_session_key: Optional[str] = "",
-        object_id: Optional[str] = "",
+        user: USER_MODEL | None = None,
+        dj_session_key: str | None = "",
+        object_id: str | None = "",
     ):
         """
         Initialize the DjangoSaver and create a UserChatSession if applicable.
@@ -317,7 +316,7 @@ async def aput_writes(
                     },
                 )
 
-    async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
+    async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database asynchronously.
 
         This method retrieves a checkpoint tuple from the database based on the
@@ -362,11 +361,11 @@ async def aget_tuple(self, config: RunnableConfig) -> Optional[CheckpointTuple]:
 
     async def alist(
         self,
-        config: Optional[RunnableConfig],
+        config: RunnableConfig | None,
         *,
-        filter: Optional[dict[str, Any]] = None,  # noqa: ARG002, A002
-        before: Optional[RunnableConfig] = None,
-        limit: Optional[int] = None,
+        filter: dict[str, Any] | None = None,  # noqa: ARG002, A002
+        before: RunnableConfig | None = None,
+        limit: int | None = None,
     ) -> AsyncGenerator[CheckpointTuple, None]:
         """List checkpoints from the database asynchronously.
 

ai_chatbots/constants.py

@@ -2,7 +2,6 @@
 
 import dataclasses
 import datetime
-from typing import Optional
 
 from named_enum import ExtendedEnum
 
@@ -57,7 +56,7 @@ class ChatbotCookie:
     name: str
     value: str
     path: str = "/"
-    max_age: Optional[datetime.datetime] = None
+    max_age: datetime.datetime | None = None
 
     def __str__(self) -> str:
         """

ai_chatbots/consumers.py

@@ -2,7 +2,6 @@
 import logging
 from abc import ABC, abstractmethod
 from http.cookies import SimpleCookie
-from typing import Optional
 from uuid import uuid4
 
 import litellm
@@ -104,9 +103,9 @@ async def assign_thread_cookies(
         self,
         user: User,
         *,
-        clear_history: Optional[bool] = False,
-        thread_id: Optional[str] = None,
-        object_id: Optional[str] = None,
+        clear_history: bool | None = False,
+        thread_id: str | None = None,
+        object_id: str | None = None,
     ) -> tuple[str, list[str]]:
         """
         Extract and update separate cookie values for logged in vs anonymous users.
@@ -220,7 +219,7 @@ async def assign_thread_cookies(
         return current_thread_id, cookies
 
     async def prepare_response(
-        self, serializer: ChatRequestSerializer, object_id_field: Optional[str] = None
+        self, serializer: ChatRequestSerializer, object_id_field: str | None = None
     ) -> tuple[str, list[str]]:
         """Prepare consumer for the API response"""
         if object_id_field:
@@ -249,9 +248,9 @@ def process_extra_state(self, data: dict) -> dict:  # noqa: ARG002
 
     async def start_response(
         self,
-        thread_id: Optional[str] = None,
-        status: Optional[int] = HTTP_200_OK,
-        cookies: Optional[list[str]] = None,
+        thread_id: str | None = None,
+        status: int | None = HTTP_200_OK,
+        cookies: list[str] | None = None,
     ):
         headers = (
             [
@@ -507,7 +506,7 @@ def process_extra_state(self, data: dict) -> dict:
     def prepare_response(
         self,
         serializer: SyllabusChatRequestSerializer,
-        object_id_field: Optional[str] = None,
+        object_id_field: str | None = None,
     ) -> tuple[str, list[str]]:
         """Set the course id as the default object id field"""
         object_id_field = object_id_field or "course_id"
@@ -620,7 +619,7 @@ def create_chatbot(
     def prepare_response(
         self,
         serializer: TutorChatRequestSerializer,
-        object_id_field: Optional[str] = None,
+        object_id_field: str | None = None,
     ) -> tuple[str, list[str]]:
         """Set the edx_module_id as the default object id field"""
         object_id_field = object_id_field or "edx_module_id"
@@ -673,7 +672,7 @@ def create_chatbot(
     def prepare_response(
         self,
         serializer: TutorChatRequestSerializer,
-        object_id_field: Optional[str] = None,
+        object_id_field: str | None = None,
     ) -> tuple[str, list[str]]:
         """Set the edx_module_id as the default object id field"""
         object_id_field = "object_id"
@@ -752,7 +751,7 @@ def process_extra_state(self, data: dict) -> dict:
     def prepare_response(
         self,
         serializer: VideoGPTRequestSerializer,
-        object_id_field: Optional[str] = None,
+        object_id_field: str | None = None,
     ) -> tuple[str, list[str]]:
         """Set the problem code as the default object id field"""
         object_id_field = object_id_field or "transcript_asset_id"

ai_chatbots/consumers_test.py

@@ -1017,7 +1017,7 @@ async def test_anonymous_user_login_session_association(  # noqa: PLR0913
     for i in range(3):
         anon_consumer = anonymous_consumer_setup(test_session_key)
         payload = {
-            "message": f"Anonymous question {i+1}",
+            "message": f"Anonymous question {i + 1}",
             "course_id": "MITx+6.00.1x",
         }
         await anon_consumer.handle(json.dumps(payload))