gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

revert changes to LoaderEngine and LoaderInterface

Browse source
This commit is contained in:
Daulet Amirkhanov 2025-10-20 22:00:13 +01:00
parent f7c2187ce7
commit 1213a3a4cb
2 changed files with 6 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
cognee/infrastructure/loaders/LoaderEngine.py
View file

@ -64,7 +64,7 @@ class LoaderEngine:
return True return True
def get_loader( def get_loader(
self, data_item_path: str, preferred_loaders: List[str] = None self, file_path: str, preferred_loaders: List[str] = None
) -> Optional[LoaderInterface]: ) -> Optional[LoaderInterface]:
""" """
Get appropriate loader for a file. Get appropriate loader for a file.
@ -76,37 +76,24 @@ class LoaderEngine:
Returns: Returns:
LoaderInterface that can handle the file, or None if not found LoaderInterface that can handle the file, or None if not found
""" """
is_url = data_item_path.startswith(("http://", "https://"))
if is_url: file_info = filetype.guess(file_path)
extension = None
mime_type = None
else:
file_info = filetype.guess(data_item_path)
extension = file_info.extension if file_info else None
mime_type = file_info.mime if file_info else None
# Try preferred loaders first # Try preferred loaders first
if preferred_loaders: if preferred_loaders:
for loader_name in preferred_loaders: for loader_name in preferred_loaders:
if loader_name in self._loaders: if loader_name in self._loaders:
loader = self._loaders[loader_name] loader = self._loaders[loader_name]
if loader.can_handle( if loader.can_handle(extension=file_info.extension, mime_type=file_info.mime):
extension=extension,
mime_type=mime_type,
data_item_path=data_item_path,
): # TODO: I'd like to refactor this to be just one argument and let loaders get file_info inside, but I'll keep that until review time
return loader return loader
else: else:
logger.info(f"Skipping {loader_name}: Preferred Loader not registered") logger.info(f"Skipping {loader_name}: Preferred Loader not registered")
# Try default priority order # Try default priority order
for loader_name in ( for loader_name in self.default_loader_priority:
self.default_loader_priority
): # TODO: I'm in favor of adding WebUrlLoader to defaults, but keeping it for review
if loader_name in self._loaders: if loader_name in self._loaders:
loader = self._loaders[loader_name] loader = self._loaders[loader_name]
if loader.can_handle(extension=extension, mime_type=mime_type): if loader.can_handle(extension=file_info.extension, mime_type=file_info.mime):
return loader return loader
else: else:
logger.info( logger.info(

4
cognee/infrastructure/loaders/LoaderInterface.py
View file

@ -44,9 +44,7 @@ class LoaderInterface(ABC):
pass pass
@abstractmethod @abstractmethod
def can_handle( def can_handle(self, extension: str, mime_type: str) -> bool:
self, extension: str, mime_type: str, data_item_path: str = None
) -> bool: # TODO: Temporarily set this to default to None so that I don't update other loaders unnecessarily yet, see TODO in LoaderEngine.py
""" """
Check if this loader can handle the given file. Check if this loader can handle the given file.