prisma-ingest 0.5.0

Python client for the Unergy Prisma document ingestion API

prisma-ingest

Python client for the Unergy Prisma document ingestion API.

Anyone can install this package, but it is only useful with a valid Unergy Prisma base_url and api_key — without them the client has nothing to talk to.

Installation

# pip
pip install prisma-ingest

# uv
uv add prisma-ingest

For Django projects, install with the optional Django extra:

pip install "prisma-ingest[django]"

# uv
uv add "prisma-ingest[django]"

Quick start

Synchronous

from prisma_ingest import PrismaIngestClient

with PrismaIngestClient(
    source="my-source",
    base_url="https://example.com",
    api_key="your-api-key",
) as client:
    doc = client.ingest(
        bucket_name="my-docs",
        key="path/informe-tecnico.pdf",
        file_type="pdf",
        uploaded_by="jcogollo"
    )
    print(doc.id, doc.status)  # 42 pending
    
    # Request LLM-based field extraction once indexed
    data = client.extract(
        bucket_name="my-docs",
        key="path/archivo.pdf",
        uploaded_by="user1",
        prompt="Your prompt here",
        schema={"type": "object", "properties": {"name": {"type": "string"}}}
    )
    if data:
        print(data)  # {"name": "Data A", ...}
    else:
        print("Document not yet indexed, try again later")

Async

import asyncio
from prisma_ingest import AsyncPrismaIngestClient

async def main():
    async with AsyncPrismaIngestClient(
        source="my-source",
        base_url="https://example.com",
        api_key="your-api-key",
    ) as client:
        doc = await client.ingest(
            bucket_name="my-docs",
            key="path/archivo.pdf",
            file_type="pdf",
            uploaded_by="user1"
        )
        print(doc.id, doc.status)
        
        # Request LLM-based field extraction
        data = await client.extract(
            bucket_name="my-docs",
            key="path/archivo.pdf",
            uploaded_by="user1",
            prompt="Your prompt here"
        )
        print(data or "Not yet indexed")

asyncio.run(main())

API reference

PrismaIngestClient(source, base_url, api_key, timeout=30.0)

Method	Description
ingest(bucket_name, key, file_type, uploaded_by)	Queue a document for ingestion. Returns a Document with status="pending". Processing is async (Celery).
extract(bucket_name, key, uploaded_by, prompt, schema=None)	Request LLM-based field extraction on an indexed document. Returns extracted data dict or None if not yet indexed.
get_document(document_id)	Fetch a document by ID, including its chunks once indexed.
list_documents()	Return all documents visible to this API key.
close()	Release the underlying HTTP connection pool. Use as context manager to close automatically.

AsyncPrismaIngestClient exposes the same methods as await-able coroutines.

Document

Field	Type	Description
id	int	Primary key
bucket	str	B2 bucket name
key	str	File path within the bucket
source	str	Originating service label
status	str	pending / processing / indexed / failed
error_message	str | None	Set when status="failed"
owner	int | None	User ID if document is user-scoped
created_at	str	ISO-8601 timestamp
updated_at	str	ISO-8601 timestamp
chunks	list[DocumentChunk]	Populated once indexed

Convenience properties: .is_indexed, .is_pending, .is_failed.

Exceptions

All exceptions inherit from PrismaIngestError.

Exception	When raised
AuthenticationError	401 / 403 — invalid or missing X-Api-Key
ValidationError	400 / 422 — payload rejected by the server
ServerError	5xx response
IngestError	Any other unexpected HTTP status

from prisma_ingest import PrismaIngestClient, AuthenticationError, ValidationError

try:
    doc = client.ingest(bucket_name="x", key="y", source="z")
except AuthenticationError:
    print("Check your API key")
except ValidationError as e:
    print("Bad request:", e)

Django integration

IngestFileField is a drop-in replacement for FileField that automatically queues files for ingestion whenever they are set or changed on a model instance.

1. Configure settings.py

PRISMA_INGEST = {
    "BASE_URL": "https://example.com",
    "API_KEY": "your-api-key",
    "SOURCE": "my-app",
    "BUCKET_NAME": "my-docs",          # global default
    "GET_UPLOADED_BY": "myapp.middleware.get_current_user",  # optional dotted path
    "TIMEOUT": 30.0,                       # optional, default 30 s
}

GET_UPLOADED_BY must be a dotted import path to a zero-argument callable that returns an object with a .username attribute (or None). If omitted, uploaded_by defaults to "unknown".

2. Replace FileField in your models

from prisma_ingest.django import IngestFileField

class Contract(models.Model):
    # uses the global BUCKET_NAME
    document = IngestFileField(upload_to="contracts/")

    # overrides bucket for this field only
    attachment = IngestFileField(upload_to="attachments/", bucket_name="other-bucket")

Every time Django saves a new or changed file, IngestFileField calls client.ingest() with the file path and the user resolved by GET_UPLOADED_BY.

The HTTP client is created once per process (lazy, cached). Non-Django code that imports prisma_ingest is unaffected — django.py is never loaded unless explicitly imported.

Disabling ingestion

Set the PRISMA_ENABLED environment variable to 0, false, no, or off to disable automatic ingestion (e.g. in tests or local development). It's read on every save, so it can be toggled at runtime. Ingestion is enabled by default.

Development

git clone https://github.com/unergy-io/prisma-ingest.git
cd prisma-ingest
uv venv && uv sync