From da1a2a1b0059bb035ea64927a7beed77766862e2 Mon Sep 17 00:00:00 2001 From: Pierre Houdyer Date: Mon, 23 Mar 2026 10:22:39 +0100 Subject: [PATCH 01/10] feat(connectors): add elasticsearch optional dependency --- pyproject.toml | 3 +++ requirements.txt | 1 + 2 files changed, 4 insertions(+) diff --git a/pyproject.toml b/pyproject.toml index 2c0577d..c3eab8d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,6 +68,9 @@ sharepoint = [ confluence = [ "atlassian-python-api>=3.41", ] +elasticsearch = [ + "elasticsearch>=8.0", +] [tool.pytest.ini_options] DJANGO_SETTINGS_MODULE = "score.settings" diff --git a/requirements.txt b/requirements.txt index 22492e6..c30afc5 100644 --- a/requirements.txt +++ b/requirements.txt @@ -59,6 +59,7 @@ whitenoise>=6.7 # msal>=1.28 # office365-rest-python-client>=2.5 # atlassian-python-api>=3.41 +# elasticsearch>=8.0 # --- Dev --- # pytest>=8.0 From 3c8f7df322bbd83bb3f616ae3a6ceba711eaf58f Mon Sep 17 00:00:00 2001 From: Pierre Houdyer Date: Mon, 23 Mar 2026 10:23:01 +0100 Subject: [PATCH 02/10] feat(connectors): add Elasticsearch to ConnectorType choices --- .../0003_add_elasticsearch_connector_type.py | 18 ++++++++++++++++++ connectors/models.py | 1 + 2 files changed, 19 insertions(+) create mode 100644 connectors/migrations/0003_add_elasticsearch_connector_type.py diff --git a/connectors/migrations/0003_add_elasticsearch_connector_type.py b/connectors/migrations/0003_add_elasticsearch_connector_type.py new file mode 100644 index 0000000..19179ca --- /dev/null +++ b/connectors/migrations/0003_add_elasticsearch_connector_type.py @@ -0,0 +1,18 @@ +# Generated by Django 5.1.15 on 2026-03-23 09:22 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('connectors', '0002_add_project'), + ] + + operations = [ + migrations.AlterField( + model_name='connectorconfig', + name='connector_type', + field=models.CharField(choices=[('sharepoint', 'SharePoint'), ('confluence', 'Confluence'), ('elasticsearch', 'Elasticsearch'), ('generic', 'Générique (Fichier/HTTP)')], max_length=20), + ), + ] diff --git a/connectors/models.py b/connectors/models.py index 114e883..3255d34 100644 --- a/connectors/models.py +++ b/connectors/models.py @@ -18,6 +18,7 @@ class ConnectorConfig(ProjectScopedModel): class ConnectorType(models.TextChoices): SHAREPOINT = "sharepoint", _("SharePoint") CONFLUENCE = "confluence", _("Confluence") + ELASTICSEARCH = "elasticsearch", _("Elasticsearch") GENERIC = "generic", _("Générique (Fichier/HTTP)") id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) From 39a5a1c84492edc223509f5717784108ea13db70 Mon Sep 17 00:00:00 2001 From: Pierre Houdyer Date: Mon, 23 Mar 2026 10:24:26 +0100 Subject: [PATCH 03/10] feat(connectors): implement Elasticsearch connector Supports API key, basic auth, and bearer token authentication. Uses search_after + PIT for efficient pagination with fallback to helpers.scan(). Configurable field mapping, optional query filter, and Elastic Cloud via cloud_id. --- connectors/__init__.py | 5 + connectors/elasticsearch.py | 331 ++++++++++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+) create mode 100644 connectors/elasticsearch.py diff --git a/connectors/__init__.py b/connectors/__init__.py index 8c43c57..13e5fe4 100644 --- a/connectors/__init__.py +++ b/connectors/__init__.py @@ -1,2 +1,7 @@ # Import connector modules so they register via @register_connector from . import generic # noqa: F401 + +try: + from . import elasticsearch # noqa: F401 +except ImportError: + pass # elasticsearch package not installed diff --git a/connectors/elasticsearch.py b/connectors/elasticsearch.py new file mode 100644 index 0000000..5cf5300 --- /dev/null +++ b/connectors/elasticsearch.py @@ -0,0 +1,331 @@ +""" +Elasticsearch connector. + +Requires: pip install elasticsearch>=8.0 (or pip install score[elasticsearch]) + +Config keys: + - hosts: Elasticsearch URL(s), e.g. "https://localhost:9200" (required) + - cloud_id: Elastic Cloud ID (alternative to hosts) + - index: Index name or pattern to read from (required) + - auth_method: "api_key" | "basic_auth" | "bearer_token" (default: "api_key") + - username: Username for basic_auth + - verify_certs: Whether to verify TLS certificates (default: True) + - ca_certs: Path to CA bundle for TLS verification + - content_field: Document field containing the main text (default: "content") + - title_field: Document field containing the title (default: "title") + - author_field: Document field containing the author (default: "author") + - date_field: Document field containing the modification date (default: "updated_at") + - query: Optional Elasticsearch query DSL (JSON) to filter documents + - batch_size: Number of documents per scroll/search_after page (default: 500) + +Credential: API key, password, or bearer token (via get_secret / credential_ref). +""" + +import json +import logging +from datetime import datetime + +from .base import BaseConnector, RawDocument, register_connector + +logger = logging.getLogger(__name__) + +# Fields to request from Elasticsearch by default (if _source filtering is used) +_META_FIELDS = ("_id", "_version", "_seq_no", "_primary_term") + + +def _parse_datetime(value) -> datetime | None: + """Try to parse a datetime from an Elasticsearch field value.""" + if value is None: + return None + if isinstance(value, datetime): + return value + if isinstance(value, (int, float)): + # Epoch millis (Elasticsearch default for date fields) + try: + return datetime.utcfromtimestamp(value / 1000) + except (ValueError, OSError): + return None + if isinstance(value, str): + for fmt in ( + "%Y-%m-%dT%H:%M:%S.%fZ", + "%Y-%m-%dT%H:%M:%SZ", + "%Y-%m-%dT%H:%M:%S.%f%z", + "%Y-%m-%dT%H:%M:%S%z", + "%Y-%m-%dT%H:%M:%S", + "%Y-%m-%d", + ): + try: + return datetime.strptime(value, fmt) + except ValueError: + continue + # Last resort: fromisoformat + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return None + + +@register_connector("elasticsearch") +class ElasticsearchConnector(BaseConnector): + """Connect to an Elasticsearch cluster and retrieve documents from an index.""" + + def __init__(self, config: dict, credential: str = ""): + super().__init__(config, credential) + + # Connection + self._hosts = config.get("hosts", "") + self._cloud_id = config.get("cloud_id", "") + self._index = config.get("index", "") + self._verify_certs = config.get("verify_certs", True) + if isinstance(self._verify_certs, str): + self._verify_certs = self._verify_certs.lower() not in ("false", "0", "no") + self._ca_certs = config.get("ca_certs", "") + + # Authentication + self._auth_method = config.get("auth_method", "api_key") + self._username = config.get("username", "") + + # Field mapping + self._content_field = config.get("content_field", "content") + self._title_field = config.get("title_field", "title") + self._author_field = config.get("author_field", "author") + self._date_field = config.get("date_field", "updated_at") + + # Query filter + query_raw = config.get("query", "") + self._query = self._parse_query(query_raw) + + # Pagination + self._batch_size = int(config.get("batch_size", 500)) + + self._client = None + + @staticmethod + def _parse_query(query_raw) -> dict: + """Parse a query from config (could be dict or JSON string).""" + if not query_raw: + return {"match_all": {}} + if isinstance(query_raw, dict): + return query_raw + try: + return json.loads(query_raw) + except (json.JSONDecodeError, TypeError): + logger.warning("Invalid Elasticsearch query, falling back to match_all: %s", query_raw) + return {"match_all": {}} + + def _get_client(self): + """Lazily create and return the Elasticsearch client.""" + if self._client is not None: + return self._client + + try: + from elasticsearch import Elasticsearch + except ImportError: + raise ImportError("Install elasticsearch extras: pip install score[elasticsearch]") + + kwargs = {} + + # TLS + kwargs["verify_certs"] = self._verify_certs + if not self._verify_certs: + kwargs["ssl_show_warn"] = False + if self._ca_certs: + kwargs["ca_certs"] = self._ca_certs + + # Authentication + secret = self.credential + if self._auth_method == "api_key": + if secret: + kwargs["api_key"] = secret + elif self._auth_method == "basic_auth": + if self._username and secret: + kwargs["basic_auth"] = (self._username, secret) + elif self._auth_method == "bearer_token": + if secret: + kwargs["bearer_auth"] = secret + + # Connection target + if self._cloud_id: + kwargs["cloud_id"] = self._cloud_id + self._client = Elasticsearch(**kwargs) + else: + hosts = [h.strip() for h in self._hosts.split(",") if h.strip()] + if not hosts: + raise ValueError("Elasticsearch connector requires 'hosts' or 'cloud_id' in config") + self._client = Elasticsearch(hosts, **kwargs) + + return self._client + + def test_connection(self) -> bool: + """Verify connection by pinging the cluster and checking the index exists.""" + try: + client = self._get_client() + if not client.ping(): + logger.warning("Elasticsearch ping failed") + return False + if self._index and not client.indices.exists(index=self._index): + logger.warning("Elasticsearch index '%s' does not exist", self._index) + return False + return True + except Exception as e: + logger.warning("Elasticsearch connection test failed: %s", e) + return False + + def list_documents(self) -> list[dict]: + """ + List all documents in the configured index using search_after + PIT + for consistent, memory-efficient pagination. + """ + client = self._get_client() + + if not self._index: + raise ValueError("Elasticsearch connector requires an 'index' config key") + + # Determine which source fields to fetch for listing (lightweight) + source_fields = [self._title_field, self._author_field, self._date_field] + source_fields = [f for f in source_fields if f] + + docs = [] + pit = None + + try: + # Open a Point in Time for consistent reads + pit_resp = client.open_point_in_time(index=self._index, keep_alive="2m") + pit_id = pit_resp["id"] + pit = {"id": pit_id, "keep_alive": "2m"} + + search_after = None + + while True: + body = { + "size": self._batch_size, + "query": self._query, + "sort": [{"_shard_doc": "asc"}], + "_source": source_fields, + "pit": pit, + "track_total_hits": False, + } + if search_after is not None: + body["search_after"] = search_after + + resp = client.search(body=body) + hits = resp["hits"]["hits"] + if not hits: + break + + # Update PIT id (may change between requests) + pit["id"] = resp.get("pit_id", pit["id"]) + + for hit in hits: + source = hit.get("_source", {}) + version = str(hit.get("_version", hit.get("_seq_no", ""))) + + docs.append( + { + "source_id": hit["_id"], + "title": source.get(self._title_field, hit["_id"]), + "source_version": version, + "source_modified_at": source.get(self._date_field, ""), + "author": source.get(self._author_field, ""), + "content_type": "text/plain", + } + ) + + search_after = hits[-1]["sort"] + + except Exception: + # If PIT is not supported (e.g. older ES), fall back to scroll via helpers.scan + logger.info("PIT not available, falling back to helpers.scan()") + docs = self._list_documents_scan(client, source_fields) + finally: + if pit: + try: + client.close_point_in_time(id=pit["id"]) + except Exception: + pass + + return docs + + def _list_documents_scan(self, client, source_fields: list[str]) -> list[dict]: + """Fallback: list documents using helpers.scan() (scroll API).""" + from elasticsearch.helpers import scan + + docs = [] + for hit in scan( + client, + index=self._index, + query={"query": self._query, "_source": source_fields}, + scroll="2m", + size=self._batch_size, + preserve_order=False, + ): + source = hit.get("_source", {}) + version = str(hit.get("_version", hit.get("_seq_no", ""))) + docs.append( + { + "source_id": hit["_id"], + "title": source.get(self._title_field, hit["_id"]), + "source_version": version, + "source_modified_at": source.get(self._date_field, ""), + "author": source.get(self._author_field, ""), + "content_type": "text/plain", + } + ) + return docs + + def fetch_document(self, source_id: str) -> RawDocument: + """Fetch full document content by _id.""" + client = self._get_client() + doc = client.get(index=self._index, id=source_id) + + source = doc.get("_source", {}) + version = str(doc.get("_version", doc.get("_seq_no", ""))) + + # Extract content — try the configured field, fall back to concatenating all text fields + content = source.get(self._content_field, "") + if not content: + content = self._extract_text_from_source(source) + + title = source.get(self._title_field, source_id) + author = source.get(self._author_field, "") + modified_at = _parse_datetime(source.get(self._date_field)) + + # Determine content type + content_type = "text/plain" + if isinstance(content, str) and content.strip().startswith("<"): + content_type = "text/html" + + # Build a stable source_url + hosts = self._hosts or "" + base_url = hosts.split(",")[0].strip().rstrip("/") if hosts else "" + source_url = f"{base_url}/{self._index}/_doc/{source_id}" if base_url else "" + + return RawDocument( + source_id=source_id, + title=title, + content=content, + content_type=content_type, + source_url=source_url, + author=author, + doc_type="elasticsearch_doc", + source_version=version, + source_modified_at=modified_at, + metadata={ + "index": doc.get("_index", self._index), + "seq_no": doc.get("_seq_no"), + "primary_term": doc.get("_primary_term"), + }, + ) + + @staticmethod + def _extract_text_from_source(source: dict) -> str: + """ + Concatenate all string fields from the document source as a fallback + when the configured content_field is empty or missing. + """ + parts = [] + for key, value in source.items(): + if isinstance(value, str) and len(value) > 20: + parts.append(value) + return "\n\n".join(parts) From 073e843c86484c59afb44a882ef4fb96783db266 Mon Sep 17 00:00:00 2001 From: Pierre Houdyer Date: Mon, 23 Mar 2026 10:27:10 +0100 Subject: [PATCH 04/10] feat(ui): add Elasticsearch config panels and icon to connector templates Adds connection form fields for hosts, cloud_id, index, auth method, username, content/title field mapping, and TLS verification. Uses a yellow accent icon (Elasticsearch brand color). --- .../connectors/_connector_cards.html | 53 ++++++++++++++++++- connectors/templates/connectors/create.html | 50 +++++++++++++++++ connectors/templates/connectors/list.html | 2 + 3 files changed, 104 insertions(+), 1 deletion(-) diff --git a/connectors/templates/connectors/_connector_cards.html b/connectors/templates/connectors/_connector_cards.html index c469c19..d8c8dc3 100644 --- a/connectors/templates/connectors/_connector_cards.html +++ b/connectors/templates/connectors/_connector_cards.html @@ -47,6 +47,8 @@

{% trans "Connecteurs" %}

{% elif conn.connector_type == 'confluence' %} + {% elif conn.connector_type == 'elasticsearch' %} + {% else %} {% endif %} @@ -133,6 +135,8 @@

{% trans "Connecteurs" %}

{% elif val == 'confluence' %} + {% elif val == 'elasticsearch' %} + {% else %} {% endif %} @@ -140,7 +144,7 @@

{% trans "Connecteurs" %}

{{ label }}
- {% if val == 'sharepoint' %}{% trans "Bibliothèques SharePoint Online" %}{% elif val == 'confluence' %}{% trans "Espaces Confluence Cloud" %}{% else %}{% trans "Fichiers locaux ou URL" %}{% endif %} + {% if val == 'sharepoint' %}{% trans "Bibliothèques SharePoint Online" %}{% elif val == 'confluence' %}{% trans "Espaces Confluence Cloud" %}{% elif val == 'elasticsearch' %}{% trans "Index Elasticsearch" %}{% else %}{% trans "Fichiers locaux ou URL" %}{% endif %}
@@ -226,6 +230,51 @@

{% trans "Connecteurs" %}

+ + + + + +