Commit 29f904db authored by Michael Poluektov's avatar Michael Poluektov
Browse files

remove List imports

parent 038fc48a
......@@ -438,7 +438,7 @@ def transcribe(
)
def get_available_models() -> List[dict]:
def get_available_models() -> list[dict]:
if app.state.config.TTS_ENGINE == "openai":
return [{"id": "tts-1"}, {"id": "tts-1-hd"}]
elif app.state.config.TTS_ENGINE == "elevenlabs":
......@@ -466,7 +466,7 @@ async def get_models(user=Depends(get_verified_user)):
return {"models": get_available_models()}
def get_available_voices() -> List[dict]:
def get_available_voices() -> list[dict]:
if app.state.config.TTS_ENGINE == "openai":
return [
{"name": "alloy", "id": "alloy"},
......
......@@ -114,7 +114,7 @@ async def get_ollama_api_urls(user=Depends(get_admin_user)):
class UrlUpdateForm(BaseModel):
urls: List[str]
urls: list[str]
@app.post("/urls/update")
......@@ -646,7 +646,7 @@ def generate_ollama_embeddings(
class GenerateCompletionForm(BaseModel):
model: str
prompt: str
images: Optional[List[str]] = None
images: Optional[list[str]] = None
format: Optional[str] = None
options: Optional[dict] = None
system: Optional[str] = None
......@@ -689,12 +689,12 @@ async def generate_completion(
class ChatMessage(BaseModel):
role: str
content: str
images: Optional[List[str]] = None
images: Optional[list[str]] = None
class GenerateChatCompletionForm(BaseModel):
model: str
messages: List[ChatMessage]
messages: list[ChatMessage]
format: Optional[str] = None
options: Optional[dict] = None
template: Optional[str] = None
......@@ -772,7 +772,7 @@ class OpenAIChatMessage(BaseModel):
class OpenAIChatCompletionForm(BaseModel):
model: str
messages: List[OpenAIChatMessage]
messages: list[OpenAIChatMessage]
model_config = ConfigDict(extra="allow")
......
......@@ -33,7 +33,7 @@ from config import (
MODEL_FILTER_LIST,
AppConfig,
)
from typing import List, Optional, Literal, overload
from typing import Optional, Literal, overload
import hashlib
......@@ -89,11 +89,11 @@ async def update_config(form_data: OpenAIConfigForm, user=Depends(get_admin_user
class UrlsUpdateForm(BaseModel):
urls: List[str]
urls: list[str]
class KeysUpdateForm(BaseModel):
keys: List[str]
keys: list[str]
@app.get("/urls")
......
......@@ -13,7 +13,7 @@ import os, shutil, logging, re
from datetime import datetime
from pathlib import Path
from typing import List, Union, Sequence, Iterator, Any
from typing import Union, Sequence, Iterator, Any
from chromadb.utils.batch_utils import create_batches
from langchain_core.documents import Document
......@@ -439,7 +439,7 @@ class ChunkParamUpdateForm(BaseModel):
class YoutubeLoaderConfig(BaseModel):
language: List[str]
language: list[str]
translation: Optional[str] = None
......@@ -642,7 +642,7 @@ def query_doc_handler(
class QueryCollectionsForm(BaseModel):
collection_names: List[str]
collection_names: list[str]
query: str
k: Optional[int] = None
r: Optional[float] = None
......@@ -1021,7 +1021,7 @@ class TikaLoader:
self.file_path = file_path
self.mime_type = mime_type
def load(self) -> List[Document]:
def load(self) -> list[Document]:
with open(self.file_path, "rb") as f:
data = f.read()
......
import logging
from typing import List, Optional
from typing import Optional
import requests
from apps.rag.search.main import SearchResult, get_filtered_results
......@@ -10,7 +10,7 @@ log.setLevel(SRC_LOG_LEVELS["RAG"])
def search_brave(
api_key: str, query: str, count: int, filter_list: Optional[List[str]] = None
api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None
) -> list[SearchResult]:
"""Search using Brave's Search API and return the results as a list of SearchResult objects.
......
import logging
from typing import List, Optional
from typing import Optional
from apps.rag.search.main import SearchResult, get_filtered_results
from duckduckgo_search import DDGS
from config import SRC_LOG_LEVELS
......@@ -9,7 +9,7 @@ log.setLevel(SRC_LOG_LEVELS["RAG"])
def search_duckduckgo(
query: str, count: int, filter_list: Optional[List[str]] = None
query: str, count: int, filter_list: Optional[list[str]] = None
) -> list[SearchResult]:
"""
Search using DuckDuckGo's Search API and return the results as a list of SearchResult objects.
......@@ -18,7 +18,7 @@ def search_duckduckgo(
count (int): The number of results to return
Returns:
List[SearchResult]: A list of search results
list[SearchResult]: A list of search results
"""
# Use the DDGS context manager to create a DDGS object
with DDGS() as ddgs:
......
import json
import logging
from typing import List, Optional
from typing import Optional
import requests
from apps.rag.search.main import SearchResult, get_filtered_results
......@@ -15,7 +15,7 @@ def search_google_pse(
search_engine_id: str,
query: str,
count: int,
filter_list: Optional[List[str]] = None,
filter_list: Optional[list[str]] = None,
) -> list[SearchResult]:
"""Search using Google's Programmable Search Engine API and return the results as a list of SearchResult objects.
......
......@@ -17,7 +17,7 @@ def search_jina(query: str, count: int) -> list[SearchResult]:
count (int): The number of results to return
Returns:
List[SearchResult]: A list of search results
list[SearchResult]: A list of search results
"""
jina_search_endpoint = "https://s.jina.ai/"
headers = {
......
import logging
import requests
from typing import List, Optional
from typing import Optional
from apps.rag.search.main import SearchResult, get_filtered_results
from config import SRC_LOG_LEVELS
......@@ -14,9 +14,9 @@ def search_searxng(
query_url: str,
query: str,
count: int,
filter_list: Optional[List[str]] = None,
filter_list: Optional[list[str]] = None,
**kwargs,
) -> List[SearchResult]:
) -> list[SearchResult]:
"""
Search a SearXNG instance for a given query and return the results as a list of SearchResult objects.
......@@ -31,10 +31,10 @@ def search_searxng(
language (str): Language filter for the search results; e.g., "en-US". Defaults to an empty string.
safesearch (int): Safe search filter for safer web results; 0 = off, 1 = moderate, 2 = strict. Defaults to 1 (moderate).
time_range (str): Time range for filtering results by date; e.g., "2023-04-05..today" or "all-time". Defaults to ''.
categories: (Optional[List[str]]): Specific categories within which the search should be performed, defaulting to an empty string if not provided.
categories: (Optional[list[str]]): Specific categories within which the search should be performed, defaulting to an empty string if not provided.
Returns:
List[SearchResult]: A list of SearchResults sorted by relevance score in descending order.
list[SearchResult]: A list of SearchResults sorted by relevance score in descending order.
Raise:
requests.exceptions.RequestException: If a request error occurs during the search process.
......
import json
import logging
from typing import List, Optional
from typing import Optional
import requests
from apps.rag.search.main import SearchResult, get_filtered_results
......@@ -11,7 +11,7 @@ log.setLevel(SRC_LOG_LEVELS["RAG"])
def search_serper(
api_key: str, query: str, count: int, filter_list: Optional[List[str]] = None
api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None
) -> list[SearchResult]:
"""Search using serper.dev's API and return the results as a list of SearchResult objects.
......
import json
import logging
from typing import List, Optional
from typing import Optional
import requests
from urllib.parse import urlencode
......@@ -19,7 +19,7 @@ def search_serply(
limit: int = 10,
device_type: str = "desktop",
proxy_location: str = "US",
filter_list: Optional[List[str]] = None,
filter_list: Optional[list[str]] = None,
) -> list[SearchResult]:
"""Search using serper.dev's API and return the results as a list of SearchResult objects.
......
import json
import logging
from typing import List, Optional
from typing import Optional
import requests
from apps.rag.search.main import SearchResult, get_filtered_results
......@@ -14,7 +14,7 @@ def search_serpstack(
api_key: str,
query: str,
count: int,
filter_list: Optional[List[str]] = None,
filter_list: Optional[list[str]] = None,
https_enabled: bool = True,
) -> list[SearchResult]:
"""Search using serpstack.com's and return the results as a list of SearchResult objects.
......
......@@ -17,7 +17,7 @@ def search_tavily(api_key: str, query: str, count: int) -> list[SearchResult]:
query (str): The query to search for
Returns:
List[SearchResult]: A list of search results
list[SearchResult]: A list of search results
"""
url = "https://api.tavily.com/search"
data = {"query": query, "api_key": api_key}
......
......@@ -2,7 +2,7 @@ import os
import logging
import requests
from typing import List, Union
from typing import Union
from apps.ollama.main import (
generate_ollama_embeddings,
......@@ -142,7 +142,7 @@ def merge_and_sort_query_results(query_results, k, reverse=False):
def query_collection(
collection_names: List[str],
collection_names: list[str],
query: str,
embedding_function,
k: int,
......@@ -163,7 +163,7 @@ def query_collection(
def query_collection_with_hybrid_search(
collection_names: List[str],
collection_names: list[str],
query: str,
embedding_function,
k: int,
......@@ -411,7 +411,7 @@ class ChromaRetriever(BaseRetriever):
query: str,
*,
run_manager: CallbackManagerForRetrieverRun,
) -> List[Document]:
) -> list[Document]:
query_embeddings = self.embedding_function(query)
results = self.collection.query(
......
from pydantic import BaseModel, ConfigDict
from typing import List, Union, Optional
from typing import Union, Optional
import json
import uuid
......@@ -215,7 +215,7 @@ class ChatTable:
def get_archived_chat_list_by_user_id(
self, user_id: str, skip: int = 0, limit: int = 50
) -> List[ChatModel]:
) -> list[ChatModel]:
with get_db() as db:
all_chats = (
......@@ -233,7 +233,7 @@ class ChatTable:
include_archived: bool = False,
skip: int = 0,
limit: int = 50,
) -> List[ChatModel]:
) -> list[ChatModel]:
with get_db() as db:
query = db.query(Chat).filter_by(user_id=user_id)
if not include_archived:
......@@ -251,7 +251,7 @@ class ChatTable:
include_archived: bool = False,
skip: int = 0,
limit: int = -1,
) -> List[ChatTitleIdResponse]:
) -> list[ChatTitleIdResponse]:
with get_db() as db:
query = db.query(Chat).filter_by(user_id=user_id)
if not include_archived:
......@@ -279,8 +279,8 @@ class ChatTable:
]
def get_chat_list_by_chat_ids(
self, chat_ids: List[str], skip: int = 0, limit: int = 50
) -> List[ChatModel]:
self, chat_ids: list[str], skip: int = 0, limit: int = 50
) -> list[ChatModel]:
with get_db() as db:
all_chats = (
db.query(Chat)
......@@ -322,7 +322,7 @@ class ChatTable:
except Exception:
return None
def get_chats(self, skip: int = 0, limit: int = 50) -> List[ChatModel]:
def get_chats(self, skip: int = 0, limit: int = 50) -> list[ChatModel]:
with get_db() as db:
all_chats = (
......@@ -332,7 +332,7 @@ class ChatTable:
)
return [ChatModel.model_validate(chat) for chat in all_chats]
def get_chats_by_user_id(self, user_id: str) -> List[ChatModel]:
def get_chats_by_user_id(self, user_id: str) -> list[ChatModel]:
with get_db() as db:
all_chats = (
......@@ -342,7 +342,7 @@ class ChatTable:
)
return [ChatModel.model_validate(chat) for chat in all_chats]
def get_archived_chats_by_user_id(self, user_id: str) -> List[ChatModel]:
def get_archived_chats_by_user_id(self, user_id: str) -> list[ChatModel]:
with get_db() as db:
all_chats = (
......
from pydantic import BaseModel, ConfigDict
from typing import List, Optional
from typing import Optional
import time
import logging
......@@ -105,7 +105,7 @@ class DocumentsTable:
except Exception:
return None
def get_docs(self) -> List[DocumentModel]:
def get_docs(self) -> list[DocumentModel]:
with get_db() as db:
return [
......
from pydantic import BaseModel, ConfigDict
from typing import List, Union, Optional
from typing import Union, Optional
import time
import logging
......@@ -93,7 +93,7 @@ class FilesTable:
except Exception:
return None
def get_files(self) -> List[FileModel]:
def get_files(self) -> list[FileModel]:
with get_db() as db:
return [FileModel.model_validate(file) for file in db.query(File).all()]
......
from pydantic import BaseModel, ConfigDict
from typing import List, Union, Optional
from typing import Union, Optional
import time
import logging
......@@ -125,7 +125,7 @@ class FunctionsTable:
except Exception:
return None
def get_functions(self, active_only=False) -> List[FunctionModel]:
def get_functions(self, active_only=False) -> list[FunctionModel]:
with get_db() as db:
if active_only:
......@@ -141,7 +141,7 @@ class FunctionsTable:
def get_functions_by_type(
self, type: str, active_only=False
) -> List[FunctionModel]:
) -> list[FunctionModel]:
with get_db() as db:
if active_only:
......@@ -157,7 +157,7 @@ class FunctionsTable:
for function in db.query(Function).filter_by(type=type).all()
]
def get_global_filter_functions(self) -> List[FunctionModel]:
def get_global_filter_functions(self) -> list[FunctionModel]:
with get_db() as db:
return [
......@@ -167,7 +167,7 @@ class FunctionsTable:
.all()
]
def get_global_action_functions(self) -> List[FunctionModel]:
def get_global_action_functions(self) -> list[FunctionModel]:
with get_db() as db:
return [
FunctionModel.model_validate(function)
......
from pydantic import BaseModel, ConfigDict
from typing import List, Union, Optional
from typing import Union, Optional
from sqlalchemy import Column, String, BigInteger, Text
......@@ -83,7 +83,7 @@ class MemoriesTable:
except Exception:
return None
def get_memories(self) -> List[MemoryModel]:
def get_memories(self) -> list[MemoryModel]:
with get_db() as db:
try:
......@@ -92,7 +92,7 @@ class MemoriesTable:
except Exception:
return None
def get_memories_by_user_id(self, user_id: str) -> List[MemoryModel]:
def get_memories_by_user_id(self, user_id: str) -> list[MemoryModel]:
with get_db() as db:
try:
......
......@@ -137,7 +137,7 @@ class ModelsTable:
print(e)
return None
def get_all_models(self) -> List[ModelModel]:
def get_all_models(self) -> list[ModelModel]:
with get_db() as db:
return [ModelModel.model_validate(model) for model in db.query(Model).all()]
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment