Glean Indexing SDK

A Python SDK for building custom Glean indexing connectors. Provides base classes and utilities to create connectors that fetch data from external systems and upload to Glean's indexing APIs.

Requirements

• Python >= 3.10
• A Glean instance and an indexing API token

Installation

pip install glean-indexing-sdk

Key Concepts

Every connector has two parts:

1. DataClient — fetches raw data from your external system (API, database, files)
2. Connector — transforms that data into Glean's format and uploads it

The workflow is: fetch → transform → upload. You implement get_source_data() on your data client and transform() on your connector; the SDK handles batching and upload.

See Architecture overview for a data flow diagram and the full class hierarchy.

Quickstart

1. Set up credentials

export GLEAN_SERVER_URL="https://your-company-be.glean.com"
export GLEAN_INDEXING_API_TOKEN="your-indexing-api-token"

# Deprecated alternative: use GLEAN_INSTANCE as legacy fallback
# export GLEAN_INSTANCE="acme"

2. Build a connector

This complete example defines a data type, a data client, and a connector, then indexes everything into Glean:

from typing import List, Sequence, TypedDict

from glean.indexing.connectors import BaseConnectorDataClient, BaseDatasourceConnector
from glean.indexing.models import (
    ContentDefinition,
    CustomDatasourceConfig,
    DocumentDefinition,
    IndexingMode,
    UserReferenceDefinition,
)


class WikiPageData(TypedDict):
    id: str
    title: str
    content: str
    author: str
    created_at: str
    updated_at: str
    url: str
    tags: List[str]


class WikiDataClient(BaseConnectorDataClient[WikiPageData]):
    def __init__(self, wiki_base_url: str, api_token: str):
        self.wiki_base_url = wiki_base_url
        self.api_token = api_token

    def get_source_data(self, since=None) -> Sequence[WikiPageData]:
        # Example static data
        return [
            {
                "id": "page_123",
                "title": "Engineering Onboarding Guide",
                "content": "Welcome to the engineering team...",
                "author": "jane.smith@company.com",
                "created_at": "2024-01-15T10:00:00Z",
                "updated_at": "2024-02-01T14:30:00Z",
                "url": f"{self.wiki_base_url}/pages/123",
                "tags": ["onboarding", "engineering"],
            },
            {
                "id": "page_124",
                "title": "API Documentation Standards",
                "content": "Our standards for API documentation...",
                "author": "john.doe@company.com",
                "created_at": "2024-01-20T09:15:00Z",
                "updated_at": "2024-01-25T16:45:00Z",
                "url": f"{self.wiki_base_url}/pages/124",
                "tags": ["api", "documentation", "standards"],
            },
        ]


class CompanyWikiConnector(BaseDatasourceConnector[WikiPageData]):
    configuration: CustomDatasourceConfig = CustomDatasourceConfig(
        name="company_wiki",
        display_name="Company Wiki",
        url_regex=r"https://wiki\.company\.com/.*",
        trust_url_regex_for_view_activity=True,
        is_user_referenced_by_email=True,
    )

    def transform(self, data: Sequence[WikiPageData]) -> List[DocumentDefinition]:
        documents = []
        for page in data:
            documents.append(
                DocumentDefinition(
                    id=page["id"],
                    title=page["title"],
                    datasource=self.name,
                    view_url=page["url"],
                    body=ContentDefinition(mime_type="text/plain", text_content=page["content"]),
                    author=UserReferenceDefinition(email=page["author"]),
                    created_at=self._parse_timestamp(page["created_at"]),
                    updated_at=self._parse_timestamp(page["updated_at"]),
                    tags=page["tags"],
                )
            )
        return documents

    def _parse_timestamp(self, timestamp_str: str) -> int:
        from datetime import datetime

        dt = datetime.fromisoformat(timestamp_str.replace("Z", "+00:00"))
        return int(dt.timestamp())


data_client = WikiDataClient(wiki_base_url="https://wiki.company.com", api_token="your-wiki-token")
connector = CompanyWikiConnector(name="company_wiki", data_client=data_client)
connector.configure_datasource()
connector.index_data(mode=IndexingMode.FULL)

Connector Types

Connector	Data Client	Best For
BaseDatasourceConnector	BaseDataClient	Small-to-medium datasets that fit in memory. Wikis, knowledge bases, file systems.
BaseStreamingDatasourceConnector	BaseStreamingDataClient	Large or paginated datasets where you need to limit memory usage. Uses sync generators.
BaseAsyncStreamingDatasourceConnector	BaseAsyncStreamingDataClient	Large datasets with async APIs (aiohttp, httpx async). Non-blocking I/O.
BasePeopleConnector	—	Employee and identity data indexing.

For detailed guidance on choosing between these, see the decision matrix.

Indexing Modes

• IndexingMode.FULL — Re-indexes all documents. Use for initial loads or when you need a complete refresh.
• IndexingMode.INCREMENTAL — Only indexes documents modified since the last crawl. Use for scheduled updates to minimize API calls.

connector.index_data(mode=IndexingMode.FULL)         # full re-index
connector.index_data(mode=IndexingMode.INCREMENTAL)   # only changes since last run

Testing

The SDK ships two ways to test connectors without making real API calls. Both record everything the connector posts so you can assert on the result.

Quick path: run_connector

For most tests, pair run_connector with one of the static data clients:

from glean.indexing.testing import StaticDataClient, run_connector

result = run_connector(MyConnector("my_ds", StaticDataClient([
    {"id": "1", "title": "Doc 1", "url": "https://example.com/1"},
    {"id": "2", "title": "Doc 2", "url": "https://example.com/2"},
])))

result.assert_documents_posted(count=2, datasource="my_ds")
for doc in result.documents_posted:
    print(doc.title)

For people connectors, use assert_employees_posted. For async streaming connectors, use the async runner:

import pytest
from glean.indexing.testing import StaticAsyncStreamingDataClient, run_connector_async

@pytest.mark.asyncio
async def test_async_connector():
    result = await run_connector_async(MyAsyncConnector(
        "my_ds", StaticAsyncStreamingDataClient([...])
    ))
    result.assert_documents_posted(count=10)

Advanced: mock_glean_client

When you need to drive connector.index_data() yourself — to pass custom modes, options, or interleave multiple calls — use the context manager directly:

from glean.indexing.models import IndexingMode
from glean.indexing.testing import mock_glean_client

with mock_glean_client() as client:
    connector.configure_datasource()
    connector.index_data(mode=IndexingMode.INCREMENTAL)
    client.assert_datasource_configured(name="my_ds")
    client.assert_documents_posted(count=2)

The yielded client is also a thin facade over a MagicMock(spec=Glean), so any access through client.indexing.* works exactly like the real Speakeasy client — typos at any level fail loudly:

client.indexing.documents.bulk_index.assert_called_once()       # real method
client.indexing.documents.bluk_index                            # AttributeError

If you'd rather use a decorator, @with_mock_glean_client injects the client as the first positional argument.

Contributing

This project uses mise for toolchain management and uv for Python dependencies.

mise run setup              # create venv and install dependencies
mise run test               # run all tests
mise run lint               # run all linters (ruff, pyright, markdown-code)
mise run lint:fix           # auto-fix lint issues and format code

Next Steps

• Architecture overview — data flow diagram and component hierarchy
• Streaming connectors — sync and async streaming walkthroughs
• Advanced usage — connector selection guide, forced restart uploads