#1128 | Remove deprecated type hints from typing module (#1131)

embedchain/loaders/directory_loader.py

@@ -1,7 +1,7 @@
 import hashlib
 import logging
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from embedchain.config import AddConfig
 from embedchain.data_formatter.data_formatter import DataFormatter
@@ -15,7 +15,7 @@ from embedchain.utils.misc import detect_datatype
 class DirectoryLoader(BaseLoader):
     """Load data from a directory."""
 
-    def __init__(self, config: Optional[Dict[str, Any]] = None):
+    def __init__(self, config: Optional[dict[str, Any]] = None):
         super().__init__()
         config = config or {}
         self.recursive = config.get("recursive", True)

embedchain/loaders/discourse.py

@@ -1,7 +1,7 @@
 import hashlib
 import logging
 import time
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 import requests
 
@@ -10,7 +10,7 @@ from embedchain.utils.misc import clean_string
 
 
 class DiscourseLoader(BaseLoader):
-    def __init__(self, config: Optional[Dict[str, Any]] = None):
+    def __init__(self, config: Optional[dict[str, Any]] = None):
         super().__init__()
         if not config:
             raise ValueError(

embedchain/loaders/dropbox.py

@@ -1,6 +1,5 @@
 import hashlib
 import os
-from typing import List
 
 from dropbox.files import FileMetadata
 
@@ -29,7 +28,7 @@ class DropboxLoader(BaseLoader):
         except exceptions.AuthError as ex:
             raise ValueError("Invalid Dropbox access token. Please verify your token and try again.") from ex
 
-    def _download_folder(self, path: str, local_root: str) -> List[FileMetadata]:
+    def _download_folder(self, path: str, local_root: str) -> list[FileMetadata]:
         """Download a folder from Dropbox and save it preserving the directory structure."""
         entries = self.dbx.files_list_folder(path).entries
         for entry in entries:

embedchain/loaders/github.py

@@ -4,7 +4,7 @@ import logging
 import os
 import re
 import shlex
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from tqdm import tqdm
 
@@ -20,7 +20,7 @@ VALID_SEARCH_TYPES = set(["code", "repo", "pr", "issue", "discussion"])
 class GithubLoader(BaseLoader):
     """Load data from GitHub search query."""
 
-    def __init__(self, config: Optional[Dict[str, Any]] = None):
+    def __init__(self, config: Optional[dict[str, Any]] = None):
         super().__init__()
         if not config:
             raise ValueError(

embedchain/loaders/gmail.py

@@ -5,7 +5,7 @@ import os
 from email import message_from_bytes
 from email.utils import parsedate_to_datetime
 from textwrap import dedent
-from typing import Dict, List, Optional
+from typing import Optional
 
 from bs4 import BeautifulSoup
 
@@ -57,7 +57,7 @@ class GmailReader:
                 token.write(creds.to_json())
         return creds
 
-    def load_emails(self) -> List[Dict]:
+    def load_emails(self) -> list[dict]:
         response = self.service.users().messages().list(userId="me", q=self.query).execute()
         messages = response.get("messages", [])
 
@@ -67,7 +67,7 @@ class GmailReader:
         raw_message = self.service.users().messages().get(userId="me", id=message_id, format="raw").execute()
         return base64.urlsafe_b64decode(raw_message["raw"])
 
-    def _parse_email(self, raw_email) -> Dict:
+    def _parse_email(self, raw_email) -> dict:
         mime_msg = message_from_bytes(raw_email)
         return {
             "subject": self._get_header(mime_msg, "Subject"),
@@ -124,7 +124,7 @@ class GmailLoader(BaseLoader):
         return {"doc_id": self._generate_doc_id(query, data), "data": data}
 
     @staticmethod
-    def _process_email(email: Dict) -> str:
+    def _process_email(email: dict) -> str:
         content = BeautifulSoup(email["body"], "html.parser").get_text()
         content = clean_string(content)
         return dedent(
@@ -137,6 +137,6 @@ class GmailLoader(BaseLoader):
         )
 
     @staticmethod
-    def _generate_doc_id(query: str, data: List[Dict]) -> str:
+    def _generate_doc_id(query: str, data: list[dict]) -> str:
         content_strings = [email["content"] for email in data]
         return hashlib.sha256((query + ", ".join(content_strings)).encode()).hexdigest()

embedchain/loaders/json.py

@@ -2,7 +2,7 @@ import hashlib
 import json
 import os
 import re
-from typing import Dict, List, Union
+from typing import Union
 
 import requests
 
@@ -16,14 +16,14 @@ class JSONReader:
         pass
 
     @staticmethod
-    def load_data(json_data: Union[Dict, str]) -> List[str]:
+    def load_data(json_data: Union[dict, str]) -> list[str]:
         """Load data from a JSON structure.
 
         Args:
-            json_data (Union[Dict, str]): The JSON data to load.
+            json_data (Union[dict, str]): The JSON data to load.
 
         Returns:
-            List[str]: A list of strings representing the leaf nodes of the JSON.
+            list[str]: A list of strings representing the leaf nodes of the JSON.
         """
         if isinstance(json_data, str):
             json_data = json.loads(json_data)

embedchain/loaders/mysql.py

@@ -1,13 +1,13 @@
 import hashlib
 import logging
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from embedchain.loaders.base_loader import BaseLoader
 from embedchain.utils.misc import clean_string
 
 
 class MySQLLoader(BaseLoader):
-    def __init__(self, config: Optional[Dict[str, Any]]):
+    def __init__(self, config: Optional[dict[str, Any]]):
         super().__init__()
         if not config:
             raise ValueError(
@@ -20,7 +20,7 @@ class MySQLLoader(BaseLoader):
         self.cursor = None
         self._setup_loader(config=config)
 
-    def _setup_loader(self, config: Dict[str, Any]):
+    def _setup_loader(self, config: dict[str, Any]):
         try:
             import mysql.connector as sqlconnector
         except ImportError as e:

embedchain/loaders/notion.py

@@ -1,7 +1,7 @@
 import hashlib
 import logging
 import os
-from typing import Any, Dict, List, Optional
+from typing import Any, Optional
 
 import requests
 
@@ -15,7 +15,7 @@ class NotionDocument:
     A simple Document class to hold the text and additional information of a page.
     """
 
-    def __init__(self, text: str, extra_info: Dict[str, Any]):
+    def __init__(self, text: str, extra_info: dict[str, Any]):
         self.text = text
         self.extra_info = extra_info
 
@@ -82,7 +82,7 @@ class NotionPageLoader:
         result_lines = "\n".join(result_lines_arr)
         return result_lines
 
-    def load_data(self, page_ids: List[str]) -> List[NotionDocument]:
+    def load_data(self, page_ids: list[str]) -> list[NotionDocument]:
         """Load data from the given list of page IDs."""
         docs = []
         for page_id in page_ids:

embedchain/loaders/postgres.py

@@ -1,12 +1,12 @@
 import hashlib
 import logging
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 from embedchain.loaders.base_loader import BaseLoader
 
 
 class PostgresLoader(BaseLoader):
-    def __init__(self, config: Optional[Dict[str, Any]] = None):
+    def __init__(self, config: Optional[dict[str, Any]] = None):
         super().__init__()
         if not config:
             raise ValueError(f"Must provide the valid config. Received: {config}")
@@ -15,7 +15,7 @@ class PostgresLoader(BaseLoader):
         self.cursor = None
         self._setup_loader(config=config)
 
-    def _setup_loader(self, config: Dict[str, Any]):
+    def _setup_loader(self, config: dict[str, Any]):
         try:
             import psycopg
         except ImportError as e:

embedchain/loaders/slack.py

@@ -2,7 +2,7 @@ import hashlib
 import logging
 import os
 import ssl
-from typing import Any, Dict, Optional
+from typing import Any, Optional
 
 import certifi
 
@@ -13,7 +13,7 @@ SLACK_API_BASE_URL = "https://www.slack.com/api/"
 
 
 class SlackLoader(BaseLoader):
-    def __init__(self, config: Optional[Dict[str, Any]] = None):
+    def __init__(self, config: Optional[dict[str, Any]] = None):
         super().__init__()
 
         self.config = config if config else {}
@@ -24,7 +24,7 @@ class SlackLoader(BaseLoader):
         self.client = None
         self._setup_loader(self.config)
 
-    def _setup_loader(self, config: Dict[str, Any]):
+    def _setup_loader(self, config: dict[str, Any]):
         try:
             from slack_sdk import WebClient
         except ImportError as e: