Cookbook: Robust Directory Ingestion
On this page
Recipes for loading large, messy, real-world directories where some files will inevitably be corrupted, password-protected, or unsupported.
Recipe: Never let one bad file kill the batch
from zerda.document_loaders import DirectoryLoader, DirectoryLoaderConfig
config = DirectoryLoaderConfig(silent_errors=True)
docs = DirectoryLoader("./messy_archive/", config=config).load()
# Failures are logged as warnings and skipped, not raised
Recipe: Pre-flight sizing before committing to a full load
loader = DirectoryLoader("./large_archive/")
stats = loader.get_stats()
if stats["total_size_mb"] > 2000:
print(f"Warning: {stats['total_size_mb']} MB across {stats['total_files']} files")
print("By type:", stats["by_type"])
# Decide whether to filter, batch, or proceed
get_stats() re-scans the filesystem each time it's called — it doesn't cache the file list from a previous call, so calling it right before load() costs the same glob traversal twice. Accept this cost for the visibility it buys, or skip it once you trust a given directory's contents.
Recipe: Excluding noisy subdirectories and file patterns
from zerda.document_loaders import DirectoryLoaderConfig
config = DirectoryLoaderConfig(
exclude_patterns=[
"*.tmp", "*.bak", "~$*", # editor/office lock & backup files
"node_modules/*", "__pycache__/*", ".git/*",
"drafts/*", "archive/old_versions/*",
],
)
docs = DirectoryLoader("./project/", config=config).load()
Patterns are checked via fnmatch against both the path relative to the scanned root and the bare filename — see DirectoryLoader for the exact matching logic.
Recipe: Non-recursive, top-level-only scan
docs = DirectoryLoader("./reports/", recursive=False).load()
# Only files directly inside ./reports/, not subdirectories
Recipe: Sequential loading when order or true progressive streaming matters
Concurrent mode (the default) waits for all files before yielding any documents via lazy_load(). If you need results as soon as each file finishes (e.g. to show live progress in a UI, or to start downstream processing before the whole directory is done):
config = DirectoryLoaderConfig(use_multithreading=False)
loader = DirectoryLoader("./corpus/", config=config)
for doc in loader.lazy_load():
process_immediately(doc) # true streaming, one file's documents at a time
Recipe: Handling password-protected PDFs mixed into a directory
DirectoryLoader always constructs each loader with no extra kwargs, so it cannot supply a password automatically. Two options:
Option A — accept the failures, handle those files separately:
config = DirectoryLoaderConfig(silent_errors=True)
docs = DirectoryLoader("./mixed/", config=config).load()
# Then handle known password-protected files manually:
from zerda.document_loaders import PDFLoader
known_passwords = {"confidential.pdf": "secret123"}
for filename, pwd in known_passwords.items():
docs += PDFLoader(f"./mixed/{filename}", password=pwd).load()
Option B — subclass DirectoryLoader to inject per-file config:
from zerda.document_loaders import DirectoryLoader, PDFLoader, PDFLoaderConfig
class PasswordAwareDirectoryLoader(DirectoryLoader):
def __init__(self, *args, passwords: dict = None, **kwargs):
super().__init__(*args, **kwargs)
self.passwords = passwords or {}
def _get_loader_for_file(self, file_path):
if file_path.suffix.lower() == ".pdf" and file_path.name in self.passwords:
config = PDFLoaderConfig(password=self.passwords[file_path.name])
return PDFLoader(str(file_path), config=config)
return super()._get_loader_for_file(file_path)
docs = PasswordAwareDirectoryLoader(
"./mixed/", passwords={"confidential.pdf": "secret123"}
).load()
Recipe: Combining directory scanning with formats it doesn't cover
DirectoryLoader won't pick up .graphml/.gexf (use GraphLoader) or run OCR on scanned images even if they're .pdf (use PDFMultimodalLoader). Layer them:
from zerda.document_loaders import DirectoryLoader, GraphLoader
from pathlib import Path
standard_docs = DirectoryLoader("./corpus/", silent_errors=True).load()
graph_files = list(Path("./corpus/").rglob("*.graphml"))
graph_docs = []
for gf in graph_files:
graph_docs.extend(GraphLoader(str(gf)).load())
all_docs = standard_docs + graph_docs