ingest_file

Ingest a file document into Morphik

Sync

def ingest_file(
    file: Union[str, bytes, BinaryIO, Path],
    filename: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    use_colpali: bool = True,
) -> Document

Async

async def ingest_file(
    file: Union[str, bytes, BinaryIO, Path],
    filename: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    use_colpali: bool = True,
) -> Document

Parameters

• file (Union[str, bytes, BinaryIO, Path]): File to ingest (path string, bytes, file object, or Path)
• filename (str, optional): Name of the file
• metadata (Dict[str, Any], optional): Optional metadata dictionary
• use_colpali (bool, optional): Whether to use ColPali-style embedding model to ingest the file (slower, but significantly better retrieval accuracy for images). Defaults to True.

Typed Metadata

Use Python-native values (e.g., datetime, date, Decimal) in the metadata dict. The SDK serializes them and adds the corresponding metadata_types, so you can run the advanced filters documented in Metadata Filtering.

Returns

• Document: Metadata of the ingested document

Examples

Sync

from morphik import Morphik

db = Morphik()

doc = db.ingest_file(
    "document.pdf",
    filename="document.pdf",
    metadata={"category": "research", "owner": "alice"},
    use_colpali=True,
)

Async

from morphik import AsyncMorphik

async with AsyncMorphik() as db:
    doc = await db.ingest_file(
        "document.pdf",
        filename="document.pdf",
        metadata={"category": "research", "owner": "alice"},
        use_colpali=True,
    )

Async Processing

Document ingestion is processed asynchronously. The returned Document object contains an external_id that you can use to track the document’s processing status:

• Use get_document_status to check if processing is complete
• The document won’t appear in retrieval results until processing completes

# Check processing status
doc = db.ingest_file("document.pdf")
status = db.get_document_status(doc.external_id)
print(f"Status: {status.get('status')}")  # "processing", "completed", or "failed"