localvectordb.extractors package

File content extraction plugin system for LocalVectorDB Server.

This module provides a plugin-based architecture for text extraction from various file formats. All dependencies are optional and gracefully degrade if not available.

exception localvectordb.extractors.ZipBombError

Bases: Exception

Exception raised when a potential ZIP bomb is detected.

localvectordb.extractors.validate_zip_safety(file_content: bytes, max_decompressed_size: int = 1073741824, max_compression_ratio: int = 100, max_file_count: int = 10000) → None

Validate a ZIP file for potential ZIP bomb attacks.

This function checks for common ZIP bomb attack patterns: - Excessive decompressed size (billion laughs style) - High compression ratios indicating recursive compression - Excessive number of files (many small files attack)

Parameters:

  • file_content (bytes) – Raw ZIP file content

  • max_decompressed_size (int) – Maximum allowed total decompressed size in bytes (default 1GB)

  • max_compression_ratio (int) – Maximum allowed compression ratio (default 100:1)

  • max_file_count (int) – Maximum allowed number of files in archive (default 10,000)

Raises:

  • ZipBombError – If the ZIP file exhibits characteristics of a ZIP bomb

  • ValueError – If the file is not a valid ZIP archive

class localvectordb.extractors.ExtractionResult(text: str, success: bool = True, method: str | None = None, metadata: Dict[str, Any] | None = None, error: str | None = None)

Bases: object

Result of text extraction with metadata.

__init__(text: str, success: bool = True, method: str | None = None, metadata: Dict[str, Any] | None = None, error: str | None = None)
to_dict() → Dict[str, Any]

Convert to dictionary for API responses.

class localvectordb.extractors.BaseExtractor(max_file_size_bytes: int | None = None)

Bases: ABC

Abstract base class for file content extractors.

Initialize the extractor.

Parameters:

max_file_size_bytes (Optional[int]) – Maximum file size in bytes. If None, uses the module default MAX_FILE_SIZE_BYTES. Set to 0 to disable file size checking (not recommended for production).

__init__(max_file_size_bytes: int | None = None)

Initialize the extractor.

Parameters:

max_file_size_bytes (Optional[int]) – Maximum file size in bytes. If None, uses the module default MAX_FILE_SIZE_BYTES. Set to 0 to disable file size checking (not recommended for production).

property available: bool
property max_file_size_bytes: int

Maximum file size this extractor will process.

abstract property supported_extensions: List[str]

List of file extensions this extractor supports (with dots, e.g., [‘.pdf’]).

abstract property supported_mimetypes: List[str]

List of MIME types this extractor supports.

abstract property required_packages: List[str]

List of Python packages required for this extractor.

abstract property priority: int

Priority level for this extractor (higher = preferred). Default should be 10.

abstract property metadata_schema: dict[str, MetadataField]
can_extract(filename: str, mimetype: str | None = None) → bool

Check if this extractor can handle the given file.

Parameters:

  • filename (str) – Filename to check

  • mimetype (Optional[str]) – MIME type hint

Returns:

True if this extractor can handle the file

Return type:

bool

extract_text(file_content: bytes, filename: str, mimetype: str | None = None, **kwargs) → ExtractionResult

Extract text from file content.

Parameters:

  • file_content (bytes) – Raw file content

  • filename (str) – Original filename

  • mimetype (Optional[str]) – MIME type hint

  • kwargs – Optional keyword args passed to the extractor.

Returns:

Extraction result

Return type:

ExtractionResult

get_info() → Dict[str, Any]

Get information about this extractor.

class localvectordb.extractors.ExtractorRegistry

Bases: object

Registry for file content extractors.

classmethod register(extractor: Type[BaseExtractor]) → None

Register a new extractor.

classmethod get_extractor(name: str) → BaseExtractor | None

Get an extractor by name.

classmethod list_extractors(available_only: bool = True) → List[str]

List all registered extractors.

classmethod refresh_plugins()

Force re-discovery of plugins (useful for testing)

classmethod get_extractors_for_file(filename: str, mimetype: str | None = None) → List[BaseExtractor]

Get suitable extractors for a file, sorted by priority.

Parameters:

  • filename (str) – Filename to check

  • mimetype (Optional[str]) – MIME type hint

Returns:

List of suitable extractors, sorted by priority (highest first)

Return type:

List[BaseExtractor]

classmethod extract_text(file_content: bytes, filename: str, mimetype: str | None = None, **kwargs: Any) → ExtractionResult

Extract text using the best available extractor.

Parameters:

  • file_content (bytes) – Raw file content

  • filename (str) – Original filename

  • mimetype (Optional[str]) – MIME type hint

  • kwargs – Optional extractor-specific keyword arguments (e.g. security options) forwarded to the selected extractor’s extract_text.

Returns:

Extraction result from the best available extractor

Return type:

ExtractionResult

classmethod get_supported_formats() → Dict[str, Dict[str, Any]]

Get information about all supported formats.

localvectordb.extractors.get_extractor_registry()

Get the global extractor registry.

localvectordb.extractors.get_supported_formats() → Dict[str, Dict[str, Any]]

Get currently supported file formats.

Submodules