Source code for langchain_community.document_loaders.generic

from __future__ import annotations

from pathlib import Path
from typing import (
    TYPE_CHECKING,
    Any,
    Iterator,
    List,
    Literal,
    Optional,
    Sequence,
    Union,
)

from langchain_core.documents import Document

from langchain_community.document_loaders.base import BaseBlobParser, BaseLoader
from langchain_community.document_loaders.blob_loaders import (
    BlobLoader,
    FileSystemBlobLoader,
)
from langchain_community.document_loaders.parsers.registry import get_parser

if TYPE_CHECKING:
    from langchain_text_splitters import TextSplitter

_PathLike = Union[str, Path]

DEFAULT = Literal["default"]


[docs] class GenericLoader(BaseLoader): """Generic Document Loader. A generic document loader that allows combining an arbitrary blob loader with a blob parser. Examples: Parse a specific PDF file: .. code-block:: python from langchain_community.document_loaders import GenericLoader from langchain_community.document_loaders.parsers.pdf import PyPDFParser # Recursively load all text files in a directory. loader = GenericLoader.from_filesystem( "my_lovely_pdf.pdf", parser=PyPDFParser() ) .. code-block:: python from langchain_community.document_loaders import GenericLoader from langchain_community.document_loaders.blob_loaders import FileSystemBlobLoader loader = GenericLoader.from_filesystem( path="path/to/directory", glob="**/[!.]*", suffixes=[".pdf"], show_progress=True, ) docs = loader.lazy_load() next(docs) Example instantiations to change which files are loaded: .. code-block:: python # Recursively load all text files in a directory. loader = GenericLoader.from_filesystem("/path/to/dir", glob="**/*.txt") # Recursively load all non-hidden files in a directory. loader = GenericLoader.from_filesystem("/path/to/dir", glob="**/[!.]*") # Load all files in a directory without recursion. loader = GenericLoader.from_filesystem("/path/to/dir", glob="*") Example instantiations to change which parser is used: .. code-block:: python from langchain_community.document_loaders.parsers.pdf import PyPDFParser # Recursively load all text files in a directory. loader = GenericLoader.from_filesystem( "/path/to/dir", glob="**/*.pdf", parser=PyPDFParser() ) """ # noqa: E501
[docs] def __init__( self, blob_loader: BlobLoader, # type: ignore[valid-type] blob_parser: BaseBlobParser, ) -> None: """A generic document loader. Args: blob_loader: A blob loader which knows how to yield blobs blob_parser: A blob parser which knows how to parse blobs into documents """ self.blob_loader = blob_loader self.blob_parser = blob_parser
[docs] def lazy_load( self, ) -> Iterator[Document]: """Load documents lazily. Use this when working at a large scale.""" for blob in self.blob_loader.yield_blobs(): # type: ignore[attr-defined] yield from self.blob_parser.lazy_parse(blob)
[docs] def load_and_split( self, text_splitter: Optional[TextSplitter] = None ) -> List[Document]: """Load all documents and split them into sentences.""" raise NotImplementedError( "Loading and splitting is not yet implemented for generic loaders. " "When they will be implemented they will be added via the initializer. " "This method should not be used going forward." )
[docs] @classmethod def from_filesystem( cls, path: _PathLike, *, glob: str = "**/[!.]*", exclude: Sequence[str] = (), suffixes: Optional[Sequence[str]] = None, show_progress: bool = False, parser: Union[DEFAULT, BaseBlobParser] = "default", parser_kwargs: Optional[dict] = None, ) -> GenericLoader: """Create a generic document loader using a filesystem blob loader. Args: path: The path to the directory to load documents from OR the path to a single file to load. If this is a file, glob, exclude, suffixes will be ignored. glob: The glob pattern to use to find documents. suffixes: The suffixes to use to filter documents. If None, all files matching the glob will be loaded. exclude: A list of patterns to exclude from the loader. show_progress: Whether to show a progress bar or not (requires tqdm). Proxies to the file system loader. parser: A blob parser which knows how to parse blobs into documents, will instantiate a default parser if not provided. The default can be overridden by either passing a parser or setting the class attribute `blob_parser` (the latter should be used with inheritance). parser_kwargs: Keyword arguments to pass to the parser. Returns: A generic document loader. """ blob_loader = FileSystemBlobLoader( # type: ignore[attr-defined, misc] path, glob=glob, exclude=exclude, suffixes=suffixes, show_progress=show_progress, ) if isinstance(parser, str): if parser == "default": try: # If there is an implementation of get_parser on the class, use it. blob_parser = cls.get_parser(**(parser_kwargs or {})) except NotImplementedError: # if not then use the global registry. blob_parser = get_parser(parser) else: blob_parser = get_parser(parser) else: blob_parser = parser return cls(blob_loader, blob_parser)
[docs] @staticmethod def get_parser(**kwargs: Any) -> BaseBlobParser: """Override this method to associate a default parser with the class.""" raise NotImplementedError()