IRISDocumentStore¶

def count_documents(self) -> int:
    """
    Return the number of documents in the store.

    Returns
    -------
    int
        Total document count.

    Example
    -------
    >>> store.count_documents()
    5
    """
    cur = self._cursor()
    try:
        cur.execute(f"SELECT COUNT(*) FROM SQLUser.{self.table_name}")  # noqa: S608
        row = cur.fetchone()
        return int(row[0]) if row else 0
    finally:
        cur.close()

def filter_documents(
    self,
    filters: dict[str, Any] | None = None,
) -> list[Document]:
    """
    Return documents matching the provided filters.

    Filtering is performed **in-memory** after loading all documents
    from IRIS.  This ensures compatibility with IRIS Community Edition
    (which lacks ``JSON_VALUE`` as a native SQL function) and supports
    all Python types and operators defined by the Haystack protocol.

    Parameters
    ----------
    filters:
        Filter in legacy format ``{"field": value}`` or in the
        official Haystack format ``{"operator": ..., "conditions": ...}``.
        ``None`` returns all documents.

    Returns
    -------
    list[Document]
        Documents satisfying the filter.

    Examples
    --------
    No filter — all documents::

        store.filter_documents()

    Legacy filter::

        store.filter_documents({"category": "db"})

    Integer filter::

        store.filter_documents({"year": 2024})

    Official Haystack filter with ``AND`` and ``>=``::

        store.filter_documents({
            "operator": "AND",
            "conditions": [
                {"field": "meta.category", "operator": "==", "value": "db"},
                {"field": "meta.year",     "operator": ">=", "value": 2023},
            ],
        })
    """
    cur = self._cursor()
    try:
        cur.execute(
            f"SELECT id, content, meta, score "  # noqa: S608
            f"FROM SQLUser.{self.table_name}"
        )
        rows = cur.fetchall()
    finally:
        cur.close()

    docs = [self._row_to_document(row) for row in rows]
    if not filters:
        return docs
    # return [d for d in docs if _apply_filter(d.meta, filters)]
    return [d for d in docs if document_matches_filter(filters, d)]

def write_documents(
    self,
    documents: list[Document],
    policy: DuplicatePolicy = DuplicatePolicy.NONE,
) -> int:
    """
    Persist documents to IRIS.

    Documents with embeddings are stored using ``TO_VECTOR(?, DOUBLE)``
    for native IRIS vector type conversion.  The ``meta`` field is
    serialized with ``json.dumps(sort_keys=True)`` to guarantee
    deterministic key ordering.

    Parameters
    ----------
    documents:
        List of :class:`~haystack.Document` objects to persist.
    policy:
        Duplicate handling policy:

        - ``FAIL`` *(default)*: raise :exc:`DuplicateDocumentError`.
        - ``SKIP``: silently ignore duplicate documents.
        - ``OVERWRITE``: replace the existing document.

    Returns
    -------
    int
        Number of documents written.

    Raises
    ------
    DuplicateDocumentError
        If ``policy=FAIL`` and a duplicate document ID is found.
    ValueError
        If ``documents`` contains non-:class:`~haystack.Document` objects.

    Example
    -------
    >>> from haystack import Document
    >>> from haystack.document_stores.types import DuplicatePolicy
    >>> store.write_documents(
    ...     [Document(content="Hello IRIS!", meta={"lang": "en"})],
    ...     policy=DuplicatePolicy.OVERWRITE,
    ... )
    1
    """
    if policy == DuplicatePolicy.NONE:
        policy = DuplicatePolicy.FAIL

    written = 0
    cur = self._cursor()
    try:
        for doc in documents:
            if not isinstance(doc, Document):
                msg = f"Expected a Document object, got {type(doc).__name__!r}."
                raise ValueError(msg)
            existing = self._get_by_id(doc.id, cur)
            if existing:
                if policy == DuplicatePolicy.FAIL:
                    msg = f"Document with id '{doc.id}' already exists. Use DuplicatePolicy.SKIP or OVERWRITE."
                    raise DuplicateDocumentError(msg)
                if policy == DuplicatePolicy.SKIP:
                    logger.debug("Skipping duplicate document: %s", doc.id)
                    continue
                cur.execute(
                    f"DELETE FROM SQLUser.{self.table_name} WHERE id = ?",  # noqa: S608
                    [doc.id],
                )

            meta_str = json.dumps(doc.meta or {}, ensure_ascii=False, sort_keys=True)
            emb_str = self._embedding_to_str(doc.embedding)

            if emb_str:
                cur.execute(
                    f"INSERT INTO SQLUser.{self.table_name} "  # noqa: S608
                    f"(id, content, meta, score, embedding) "
                    f"VALUES (?, ?, ?, ?, TO_VECTOR(?, DOUBLE))",
                    [doc.id, doc.content or "", meta_str, doc.score, emb_str],
                )
            else:
                cur.execute(
                    f"INSERT INTO SQLUser.{self.table_name} "  # noqa: S608
                    f"(id, content, meta, score) VALUES (?, ?, ?, ?)",
                    [doc.id, doc.content or "", meta_str, doc.score],
                )
            written += 1

        self._conn.commit()
        logger.debug("Wrote %d document(s) to IRIS.", written)
        return written
    except Exception:
        self._conn.rollback()
        raise
    finally:
        cur.close()

def delete_documents(self, document_ids: list[str]) -> None:
    """
    Delete documents by ID.

    Accepts an empty list without error (idempotent). Non-existent
    IDs are silently ignored by IRIS.

    Parameters
    ----------
    document_ids:
        List of document IDs to remove.

    Example
    -------
    >>> store.delete_documents(["id-1", "id-2"])
    """
    if not document_ids:
        return
    placeholders = ",".join(["?"] * len(document_ids))
    cur = self._cursor()
    try:
        cur.execute(
            f"DELETE FROM SQLUser.{self.table_name} "  # noqa: S608
            f"WHERE id IN ({placeholders})",
            document_ids,
        )
        self._conn.commit()
    finally:
        cur.close()

def to_dict(self) -> dict[str, Any]:
    """
    Serialize the store for use in Haystack YAML/JSON pipelines.

    The password ``Secret`` is serialized according to the Haystack
    secret protocol — the resolved value is **never** included.

    Returns
    -------
    dict
        Serializable dictionary.

    Example
    -------
    >>> d = store.to_dict()
    >>> d["type"]
    'intersystems_iris_haystack.document_stores.document_store.IRISDocumentStore'
    """
    return default_to_dict(
        self,
        connection_string=self.connection_string.to_dict(),
        username=self.username.to_dict(),
        password=self.password.to_dict(),
        table_name=self.table_name,
        embedding_dim=self.embedding_dim,
        bm25_k1=self.bm25_k1,
        bm25_b=self.bm25_b,
        recreate_table=False,  # never re-drop on restore
    )

@classmethod
def from_dict(cls, data: dict[str, Any]) -> IRISDocumentStore:
    """
    Deserialize the store from a dictionary.

    Called automatically by Haystack when loading a pipeline from a
    YAML file.

    Parameters
    ----------
    data:
        Dictionary in the format produced by :meth:`to_dict`.

    Returns
    -------
    IRISDocumentStore
    """
    deserialize_secrets_inplace(
        data["init_parameters"],
        keys=["connection_string", "username", "password"],
    )
    return default_from_dict(cls, data)

def close(self) -> None:
    """Close the IRIS connection (idempotent)."""
    if self._conn:
        try:
            self._conn.close()
        except Exception:  # noqa: S110
            pass
        logger.debug("IRIS connection closed.")