refac: modernize imports, standardize type hints and docstrings

This commit is contained in:
Timothy Jaeryang Baek committed 2026-05-12 06:30:38 +09:00
1 parent 998c86a52b
commit a59c967d7e
16 files changed
+118 -142

No files matched your search

+3 -3
View File
@@ -1,14 +1,14 @@
from __future__ import annotations
import logging
import time
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_brave(api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None) -> list[SearchResult]:
def search_brave(api_key: str, query: str, count: int, filter_list: list[str | None] = None) -> list[SearchResult]:
"""Search using Brave's Search API and return the results as a list of SearchResult objects.
Args:
@@ -1,20 +1,20 @@
from __future__ import annotations
import logging
import urllib.request
from typing import Optional
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
from ddgs import DDGS
from ddgs.exceptions import RatelimitException
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_duckduckgo(
query: str,
count: int,
filter_list: Optional[list[str]] = None,
concurrent_requests: Optional[int] = None,
backend: Optional[str] = 'auto',
filter_list: list[str | None] = None,
concurrent_requests: int | None = None,
backend: str | None = 'auto',
) -> list[SearchResult]:
"""
Search using DuckDuckGo's Search API and return the results as a list of SearchResult objects.
@@ -1,19 +1,19 @@
from __future__ import annotations
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_google_pse(
api_key: str,
search_engine_id: str,
query: str,
count: int,
filter_list: Optional[list[str]] = None,
referer: Optional[str] = None,
filter_list: list[str | None] = None,
referer: str | None = None,
) -> list[SearchResult]:
"""Search using Google's Programmable Search Engine API and return the results as a list of SearchResult objects.
Handles pagination for counts greater than 10.
@@ -23,7 +23,7 @@ def search_google_pse(
search_engine_id (str): A Programmable Search Engine ID
query (str): The query to search for
count (int): The number of results to return (max 100, as PSE max results per query is 10 and max page is 10)
filter_list (Optional[list[str]], optional): A list of keywords to filter out from results. Defaults to None.
filter_list (list[str | None], optional): A list of keywords to filter out from results. Defaults to None.
Returns:
list[SearchResult]: A list of SearchResult objects.
@@ -1,3 +1,5 @@
from __future__ import annotations
import logging
import requests
@@ -6,7 +8,6 @@ from yarl import URL
log = logging.getLogger(__name__)
def search_jina(api_key: str, query: str, count: int, base_url: str = '') -> list[SearchResult]:
"""
Search using Jina's Search API and return the results as a list of SearchResult objects.
@@ -17,9 +18,9 @@ def search_jina(api_key: str, query: str, count: int, base_url: str = '') -> lis
base_url (str): Optional custom base URL for the Jina API
Returns:
list[SearchResult]: A list of search results
A list of SearchResult objects.
"""
jina_search_endpoint = base_url if base_url else 'https://s.jina.ai/'
jina_search_endpoint = base_url or 'https://s.jina.ai/'
headers = {
'Accept': 'application/json',
+5 -8
View File
@@ -1,13 +1,11 @@
import validators
from __future__ import annotations
from typing import Optional
from urllib.parse import urlparse
from pydantic import BaseModel
import validators
from open_webui.retrieval.web.utils import resolve_hostname
from open_webui.utils.misc import is_string_allowed
from pydantic import BaseModel
def get_filtered_results(results, filter_list):
if not filter_list:
@@ -39,8 +37,7 @@ def get_filtered_results(results, filter_list):
return filtered_results
class SearchResult(BaseModel):
link: str
title: Optional[str]
snippet: Optional[str]
title: str | None
snippet: str | None
+3 -3
View File
@@ -1,13 +1,13 @@
from __future__ import annotations
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_mojeek(api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None) -> list[SearchResult]:
def search_mojeek(api_key: str, query: str, count: int, filter_list: list[str | None] = None) -> list[SearchResult]:
"""Search using Mojeek's Search API and return the results as a list of SearchResult objects.
Args:
+5 -5
View File
@@ -1,17 +1,17 @@
from __future__ import annotations
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_searxng(
def search_searxng( # noqa: PLR0913
query_url: str,
query: str,
count: int,
filter_list: Optional[list[str]] = None,
filter_list: list[str | None] = None,
**kwargs,
) -> list[SearchResult]:
"""
@@ -28,7 +28,7 @@ def search_searxng(
language (str): Language filter for the search results; e.g., "all", "en-US", "es". Defaults to "all".
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: (list[str | None]): 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.
+3 -3
View File
@@ -1,14 +1,14 @@
from __future__ import annotations
import json
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_serper(api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None) -> list[SearchResult]:
def search_serper(api_key: str, query: str, count: int, filter_list: list[str | None] = None) -> list[SearchResult]:
"""Search using serper.dev's API and return the results as a list of SearchResult objects.
Args:
+3 -3
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import logging
from typing import Optional
from urllib.parse import urlencode
import requests
@@ -7,7 +8,6 @@ from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_serply(
api_key: str,
query: str,
@@ -16,7 +16,7 @@ def search_serply(
limit: int = 10,
device_type: str = 'desktop',
proxy_location: str = 'US',
filter_list: Optional[list[str]] = None,
filter_list: list[str | None] = None,
) -> list[SearchResult]:
"""Search using serper.dev's API and return the results as a list of SearchResult objects.
@@ -1,17 +1,17 @@
from __future__ import annotations
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_serpstack(
api_key: str,
query: str,
count: int,
filter_list: Optional[list[str]] = None,
filter_list: list[str | None] = None,
https_enabled: bool = True,
) -> list[SearchResult]:
"""Search using serpstack.com's and return the results as a list of SearchResult objects.
+4 -4
View File
@@ -1,17 +1,17 @@
from __future__ import annotations
import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
log = logging.getLogger(__name__)
def search_tavily(
api_key: str,
query: str,
count: int,
filter_list: Optional[list[str]] = None,
filter_list: list[str | None] = None,
# **kwargs,
) -> list[SearchResult]:
"""Search using Tavily's Search API and return the results as a list of SearchResult objects.
@@ -22,7 +22,7 @@ def search_tavily(
count (int): The maximum number of results to return
Returns:
list[SearchResult]: A list of search results
A list of SearchResult objects.
"""
url = 'https://api.tavily.com/search'
headers = {