LCORE-1051: Updated Optional type in LCORE sources

src/a2a_storage/context_store.py

@@ -1,6 +1,7 @@
 """Abstract base class for A2A context-to-conversation mapping storage."""
 
 from abc import ABC, abstractmethod
+from typing import Optional
 
 
 class A2AContextStore(ABC):
@@ -14,7 +15,7 @@ class A2AContextStore(ABC):
     """
 
     @abstractmethod
-    async def get(self, context_id: str) -> str | None:
+    async def get(self, context_id: str) -> Optional[str]:
         """Retrieve the conversation ID for an A2A context.
 
         Args:

src/a2a_storage/in_memory_context_store.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+from typing import Optional
 
 from a2a_storage.context_store import A2AContextStore
 
@@ -26,7 +27,7 @@ def __init__(self) -> None:
         self._lock = asyncio.Lock()
         self._initialized = True
 
-    async def get(self, context_id: str) -> str | None:
+    async def get(self, context_id: str) -> Optional[str]:
         """Retrieve the conversation ID for an A2A context.
 
         Args:

src/a2a_storage/postgres_context_store.py

@@ -1,6 +1,7 @@
 """PostgreSQL implementation of A2A context store."""
 
 import logging
+from typing import Optional
 
 from sqlalchemy import Column, String, Table, MetaData, select, delete
 from sqlalchemy.dialects.postgresql import insert as pg_insert
@@ -66,7 +67,7 @@ async def _ensure_initialized(self) -> None:
         if not self._initialized:
             await self.initialize()
 
-    async def get(self, context_id: str) -> str | None:
+    async def get(self, context_id: str) -> Optional[str]:
         """Retrieve the conversation ID for an A2A context.
 
         Args:

src/a2a_storage/sqlite_context_store.py

@@ -1,6 +1,7 @@
 """SQLite implementation of A2A context store."""
 
 import logging
+from typing import Optional
 
 from sqlalchemy import Column, String, Table, MetaData, select, delete
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker
@@ -65,7 +66,7 @@ async def _ensure_initialized(self) -> None:
         if not self._initialized:
             await self.initialize()
 
-    async def get(self, context_id: str) -> str | None:
+    async def get(self, context_id: str) -> Optional[str]:
         """Retrieve the conversation ID for an A2A context.
 
         Args:

src/a2a_storage/storage_factory.py

@@ -2,6 +2,7 @@
 
 import logging
 from urllib.parse import quote_plus
+from typing import Optional
 
 from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
 
@@ -24,9 +25,9 @@ class A2AStorageFactory:
     creates database-backed stores that share state across workers.
     """
 
-    _engine: AsyncEngine | None = None
-    _task_store: TaskStore | None = None
-    _context_store: A2AContextStore | None = None
+    _engine: Optional[AsyncEngine] = None
+    _task_store: Optional[TaskStore] = None
+    _context_store: Optional[A2AContextStore] = None
 
     @classmethod
     async def create_task_store(cls, config: A2AStateConfiguration) -> TaskStore:

src/app/database.py

@@ -1,7 +1,7 @@
 """Database engine management."""
 
 from pathlib import Path
-from typing import Any
+from typing import Any, Optional
 
 from sqlalchemy import create_engine, text
 from sqlalchemy.engine.base import Engine
@@ -14,8 +14,8 @@
 logger = get_logger(__name__)
 
 # pylint: disable=invalid-name
-engine: Engine | None = None
-session_local: sessionmaker | None = None
+engine: Optional[Engine] = None
+session_local: Optional[sessionmaker] = None
 
 
 def get_engine() -> Engine:

src/authorization/middleware.py

@@ -2,7 +2,7 @@
 
 import logging
 from functools import lru_cache, wraps
-from typing import Any, Callable, Tuple
+from typing import Any, Callable, Optional, Tuple
 
 from fastapi import HTTPException
 from starlette.requests import Request
@@ -107,7 +107,7 @@ async def _perform_authorization_check(
 
     authorized_actions = access_resolver.get_actions(user_roles)
 
-    req: Request | None = None
+    req: Optional[Request] = None
     if "request" in kwargs and isinstance(kwargs["request"], Request):
         req = kwargs["request"]
     else:

src/configuration.py

@@ -324,7 +324,7 @@ def token_usage_history(self) -> Optional[TokenUsageHistory]:
         token history is disabled, returns None.
 
         Returns:
-            TokenUsageHistory | None: The cached TokenUsageHistory instance
+            Optional[TokenUsageHistory]: The cached TokenUsageHistory instance
             when enabled, otherwise `None`.
 
         Raises:

src/models/cache_entry.py

@@ -1,5 +1,6 @@
 """Model for conversation history cache entry."""
 
+from typing import Optional
 from pydantic import BaseModel
 from models.responses import ReferencedDocument
 
@@ -21,4 +22,4 @@ class CacheEntry(BaseModel):
     model: str
     started_at: str
     completed_at: str
-    referenced_documents: list[ReferencedDocument] | None = None
+    referenced_documents: Optional[list[ReferencedDocument]] = None

src/models/config.py

@@ -1178,7 +1178,7 @@ def check_default_model_and_provider(self) -> Self:
 class ConversationHistoryConfiguration(ConfigurationBase):
     """Conversation history configuration."""
 
-    type: Literal["noop", "memory", "sqlite", "postgres"] | None = Field(
+    type: Optional[Literal["noop", "memory", "sqlite", "postgres"]] = Field(
         None,
         title="Conversation history database type",
         description="Type of database where the conversation history is to be stored.",
@@ -1298,7 +1298,7 @@ def storage_type(self) -> Literal["memory", "sqlite", "postgres"]:
     @property
     def config(
         self,
-    ) -> SQLiteDatabaseConfiguration | PostgreSQLDatabaseConfiguration | None:
+    ) -> Optional[SQLiteDatabaseConfiguration | PostgreSQLDatabaseConfiguration]:
         """Return the active storage configuration, or None for memory storage."""
         if self.sqlite is not None:
             return self.sqlite

src/models/requests.py

@@ -205,15 +205,15 @@ class QueryRequest(BaseModel):
 
     @field_validator("conversation_id")
     @classmethod
-    def check_uuid(cls, value: str | None) -> str | None:
+    def check_uuid(cls, value: Optional[str]) -> Optional[str]:
         """
         Validate that a conversation identifier matches the expected SUID format.
 
         Parameters:
-            value (str | None): Conversation identifier to validate; may be None.
+            value (Optional[str]): Conversation identifier to validate; may be None.
 
         Returns:
-            str | None: The original `value` if valid or `None` if not provided.
+            Optional[str]: The original `value` if valid or `None` if not provided.
 
         Raises:
             ValueError: If `value` is provided and does not conform to the

src/models/responses.py

@@ -30,7 +30,7 @@
 
 #     tool_name: str = Field(description="Name of the tool called")
 #     arguments: dict[str, Any] = Field(description="Arguments passed to the tool")
-#     result: dict[str, Any] | None = Field(None, description="Result from the tool")
+#     result: Optional[dict[str, Any]] = Field(None, description="Result from the tool")
 
 
 # class ToolResult(BaseModel):
@@ -321,7 +321,7 @@ class ConversationData(BaseModel):
     """
 
     conversation_id: str
-    topic_summary: str | None
+    topic_summary: Optional[str]
     last_message_timestamp: float
 
 
@@ -333,9 +333,13 @@ class ReferencedDocument(BaseModel):
         doc_title: Title of the referenced doc.
     """
 
-    doc_url: AnyUrl | None = Field(None, description="URL of the referenced document")
+    doc_url: Optional[AnyUrl] = Field(
+        None, description="URL of the referenced document"
+    )
 
-    doc_title: str | None = Field(None, description="Title of the referenced document")
+    doc_title: Optional[str] = Field(
+        None, description="Title of the referenced document"
+    )
 
 
 class QueryResponse(AbstractSuccessfulResponse):
@@ -353,7 +357,7 @@ class QueryResponse(AbstractSuccessfulResponse):
         available_quotas: Quota available as measured by all configured quota limiters.
     """
 
-    conversation_id: str | None = Field(
+    conversation_id: Optional[str] = Field(
         None,
         description="The optional conversation ID (UUID)",
         examples=["c5260aec-4d82-4370-9fdf-05cf908b3f16"],
@@ -404,12 +408,12 @@ class QueryResponse(AbstractSuccessfulResponse):
         examples=[{"daily": 1000, "monthly": 50000}],
     )
 
-    tool_calls: list[ToolCallSummary] | None = Field(
+    tool_calls: Optional[list[ToolCallSummary]] = Field(
         None,
         description="List of tool calls made during response generation",
     )
 
-    tool_results: list[ToolResultSummary] | None = Field(
+    tool_results: Optional[list[ToolResultSummary]] = Field(
         None,
         description="List of tool results",
     )
@@ -573,7 +577,7 @@ class ProviderHealthStatus(BaseModel):
         description="The health status",
         examples=["ok", "unhealthy", "not_implemented"],
     )
-    message: str | None = Field(
+    message: Optional[str] = Field(
         None,
         description="Optional message about the health status",
         examples=["All systems operational", "Llama Stack is unavailable"],
@@ -921,37 +925,37 @@ class ConversationDetails(BaseModel):
         examples=["c5260aec-4d82-4370-9fdf-05cf908b3f16"],
     )
 
-    created_at: str | None = Field(
+    created_at: Optional[str] = Field(
         None,
         description="When the conversation was created",
         examples=["2024-01-01T01:00:00Z"],
     )
 
-    last_message_at: str | None = Field(
+    last_message_at: Optional[str] = Field(
         None,
         description="When the last message was sent",
         examples=["2024-01-01T01:00:00Z"],
     )
 
-    message_count: int | None = Field(
+    message_count: Optional[int] = Field(
         None,
         description="Number of user messages in the conversation",
         examples=[42],
     )
 
-    last_used_model: str | None = Field(
+    last_used_model: Optional[str] = Field(
         None,
         description="Identification of the last model used for the conversation",
         examples=["gpt-4-turbo", "gpt-3.5-turbo-0125"],
     )
 
-    last_used_provider: str | None = Field(
+    last_used_provider: Optional[str] = Field(
         None,
         description="Identification of the last provider used for the conversation",
         examples=["openai", "gemini"],
     )
 
-    topic_summary: str | None = Field(
+    topic_summary: Optional[str] = Field(
         None,
         description="Topic summary for the conversation",
         examples=["Openshift Microservices Deployment Strategies"],

src/models/rlsapi/responses.py

@@ -1,5 +1,6 @@
 """Models for rlsapi v1 REST API responses."""
 
+from typing import Optional
 from pydantic import Field
 
 from models.config import ConfigurationBase
@@ -19,7 +20,7 @@ class RlsapiV1InferData(ConfigurationBase):
         description="Generated response text",
         examples=["To list files in Linux, use the `ls` command."],
     )
-    request_id: str | None = Field(
+    request_id: Optional[str] = Field(
         None,
         description="Unique request identifier",
         examples=["01JDKR8N7QW9ZMXVGK3PB5TQWZ"],

src/quota/token_usage_history.py

@@ -7,7 +7,7 @@
 
 import sqlite3
 from datetime import datetime
-from typing import Any
+from typing import Any, Optional
 
 import psycopg2
 
@@ -50,13 +50,13 @@ def __init__(self, configuration: QuotaHandlersConfiguration) -> None:
         """
         # store the configuration, it will be used
         # by reconnection logic later, if needed
-        self.sqlite_connection_config: SQLiteDatabaseConfiguration | None = (
+        self.sqlite_connection_config: Optional[SQLiteDatabaseConfiguration] = (
             configuration.sqlite
         )
-        self.postgres_connection_config: PostgreSQLDatabaseConfiguration | None = (
+        self.postgres_connection_config: Optional[PostgreSQLDatabaseConfiguration] = (
             configuration.postgres
         )
-        self.connection: Any | None = None
+        self.connection: Optional[Any] = None
 
         # initialize connection to DB
         self.connect()

src/utils/checks.py

@@ -87,7 +87,7 @@ def directory_check(
             raise InvalidConfigurationError(f"{desc} '{path}' is not writable")
 
 
-def import_python_module(profile_name: str, profile_path: str) -> ModuleType | None:
+def import_python_module(profile_name: str, profile_path: str) -> Optional[ModuleType]:
     """
     Import a Python module from a filesystem path and return the loaded module.
 
@@ -96,7 +96,7 @@ def import_python_module(profile_name: str, profile_path: str) -> ModuleType | N
         profile_path (str): Filesystem path to the Python source file; must end with `.py`.
 
     Returns:
-        ModuleType | None: The loaded module on success; `None` if
+        Optional[ModuleType]: The loaded module on success; `None` if
         `profile_path` does not end with `.py`, if a module spec or loader
         cannot be created, or if importing/executing the module fails.
     """