feat: add Valkey destination connector by atao2004 · Pull Request #747 · Unstructured-IO/unstructured-ingest

atao2004 · 2026-07-03T20:45:08Z

Summary

Adds a Valkey destination connector using valkey-glide that stores document chunks with vector embeddings in Valkey hashes and creates FT Search indexes with HNSW vector fields for semantic retrieval.

Why

Valkey is a high-performance, open-source key-value store (BSD-licensed fork of Redis) with first-class vector search support. This connector enables unstructured-ingest users to push chunked + embedded documents directly into Valkey for low-latency semantic retrieval without requiring a separate vector database.

Features

Stores document elements as Valkey hashes with text, metadata, and float32 vector embeddings
Creates Valkey Search (FT) index with HNSW vector field for cosine/L2/IP similarity
Supports connection via individual parameters (host, port, username, password) or URI
Supports optional TTL on uploaded keys
Uses batch pipeline writes for first upload (no index overhead)
Falls back to individual HSET for subsequent uploads when index is active (avoids GLIDE batch timeout with active vector indexes)
Maps GLIDE exceptions to unstructured-ingest error types (DestinationConnectionError, UserAuthError, WriteError)
Registers as valkey connector type in the destination registry

Testing

7 integration tests (require valkey/valkey-bundle on localhost:6379):

docker run -d --name valkey-test -p 127.0.0.1:6379:6379 valkey/valkey-bundle:latest
python -m pytest test/integration/connectors/test_valkey.py -v

Test	Description
`test_valkey_destination_upload`	Upload with vector index creation and data validation
`test_valkey_destination_with_ttl`	TTL applied to uploaded keys
`test_valkey_destination_with_uri`	Connection via URI string
`test_valkey_precheck_failure`	`DestinationConnectionError` on bad connection
`test_valkey_destination_without_embeddings`	Text/metadata upload without vectors (no index)
`test_valkey_destination_idempotent`	Re-upload same data, no duplicates
`test_valkey_destination_incremental_upload`	Different data to same index

Dependencies

valkey-glide>=2.0 (added as optional extra valkey in pyproject.toml)
numpy (already a project dependency)

Changes

unstructured_ingest/processes/connectors/valkey.py — connector implementation
unstructured_ingest/processes/connectors/__init__.py — registry entry
test/integration/connectors/test_valkey.py — integration tests
pyproject.toml — valkey extra dependency
uv.lock — lockfile update

Adds a Valkey destination connector using valkey-glide that: - Stores document chunks with vector embeddings as Valkey hashes - Creates FT Search index with HNSW vector field for semantic retrieval - Supports connection via host/port/username/password or URI - Supports TTL on uploaded keys - Uses batch writes for first upload, individual hset when index active - Maps GLIDE exceptions to unstructured-ingest error types - Removes dependency on valkey-glide-sync (async-only client) Integration tests (7 passing): - Upload with vector index creation - Upload with TTL - Upload without embeddings (no index created) - Connection via URI - Precheck failure on bad connection - Idempotent re-upload (same data, no duplicates) - Incremental upload (different data, same index) Signed-off-by: Anna Tao <annatao2004@gmail.com>

…leanup

- Add element_id and embeddings type validation (Critical #1) - Document partial batch failure semantics (Critical Unstructured-IO#2) - Use @asynccontextmanager for client lifecycle (Warning Unstructured-IO#3) - Make request_timeout configurable (Warning Unstructured-IO#4) - Default ssl=True for production safety (Warning Unstructured-IO#5) - Replace bare except with ft.list() for index check (Warning Unstructured-IO#6) - Extract _build_fields and _validate_element_id helpers (Warning Unstructured-IO#7) - Pin valkey-glide~=2.4 upper bound (Warning Unstructured-IO#8) - Add TOCTOU race comment (Warning Unstructured-IO#9) - Add cluster mode note to key_prefix description (Nit Unstructured-IO#12) - Rename _sync_create_destination to _create_destination_with_client (Nit Unstructured-IO#11) - Add valkey-glide-sync for native sync client in precheck/create_destination - Remove asyncio.run() entirely — sync hooks use sync client directly - Add tests for input validation and client cleanup Signed-off-by: Anna Tao <annatao2004@gmail.com>

…or kwargs for config - Validate embedding dimension per-element against expected index schema dimension (derived from first element). Raises WriteError on mismatch instead of silently storing misaligned vectors. - Refactor GlideClientConfiguration construction to pass all options (including credentials) via constructor kwargs instead of post-construction attribute assignment. Defensive against future immutability changes. Addresses review findings from round 2 (Jonathan-Improving). Signed-off-by: Anna Tao <annatao2004@gmail.com>

- Default port to 6379 when URI omits port (parsed.port or 6379) - Route bare raise in create_destination through error mapper - Fix import ordering in __init__.py (valkey between vectara and zendesk) - Extract duplicated test cleanup into shared _cleanup() helper - Add parametrized unit test for _map_glide_error branches - Add TTL test covering individual write path (pre-existing index) - Regenerate uv.lock to include valkey-glide-sync Signed-off-by: Anna Tao <annatao2004@gmail.com>

- Fix @requires_dependencies to check glide_sync for create_destination - Add early-return in _map_glide_error for already-translated errors - Validate embedding ndim after np.array conversion (catch nested lists) - Extract _resolve_connection_params() to DRY URI parsing - Add _delete_by_record_id: delete stale chunks before re-ingestion - Fix auth substring match: use startswith instead of 'in' - Make distance_metric configurable (COSINE/L2/IP) - Add record_id TAG field to schema + stored on each hash - Extract _build_schema() and _get_distance_metric() helpers Signed-off-by: Anna Tao <annatao2004@gmail.com>

- Use correct GLIDE FtSearchResponse format: [count, {key: {fields}, ...}] - Robust TAG escaping for all RediSearch special characters - Batch delete via single client.delete(keys_list) call - Add FtSearchOptions with 10k limit + warning for overflow - Narrow exception handling: only catch RequestError 'no such index' - Add test_valkey_destination_reingestion_deletes_stale proving happy path - Fix incremental test to use different record IDs per source document Signed-off-by: Anna Tao <annatao2004@gmail.com>

- Use Batch(is_atomic=True) for HSET+EXPIRE in _write_individual to prevent TTL-less keys on partial failure - Validate URI scheme against recognized values (valkey/valkeys/redis/rediss) and reject typos like 'vakeys://' with clear error - Change distance_metric type from str to Literal['COSINE', 'L2', 'IP'] for pydantic schema validation and IDE support - Add tests for all three: scheme rejection, valid schemes with TLS mapping, metric validation, and atomic TTL on individual write path Signed-off-by: Anna Tao <annatao2004@gmail.com>

- Critical: robust TAG escaping via regex (covers . / and all separators) - Add regression test with path-like identifier (s3://bucket/file.pdf) - Sanitize precheck error message (no credential leak) - Add 'noperm' to auth error prefix matching - Store full metadata as metadata_json for retrieval - Early return for empty data in run_data_async - Revert sub-batching (causes HNSW timeout), keep per-element atomic writes Signed-off-by: Anna Tao <annatao2004@gmail.com>

socket-security · 2026-07-03T20:45:46Z

Review the following changes in direct dependencies. Learn more about Socket for GitHub.

Diff	Package	Supply Chain Security	Vulnerability	Quality	Maintenance	License
	pypi/valkey-glide-sync@2.4.2
	pypi/valkey-glide@2.4.1

View full report

cubic-dev-ai

6 issues found across 5 files

_{Shadow auto-approve: would not auto-approve because issues were found.

Re-trigger cubic}

- Make _index_exists() call unconditional in run_data_async (P1) - Parse URI username as fallback in _resolve_connection_params (P2) - Add pagination loop in _delete_by_record_id for >10k keys (P2) - Fix import sort order in __init__.py (.valkey before .vectara) (P2) - Wrap test bodies in try/finally for guaranteed cleanup (P2) - Replace time.sleep with await asyncio.sleep in async tests (P3) Signed-off-by: Anna Tao <annatao2004@gmail.com>

cubic-dev-ai

1 issue found across 3 files (changes from recent commits).

Prompt for AI agents (unresolved issues)


Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="unstructured_ingest/processes/connectors/valkey.py">

<violation number="1" location="unstructured_ingest/processes/connectors/valkey.py:366">
P2: Text-only uploads can fail before any HSET writes because `run_data_async` now always executes `FT.LIST` via `_index_exists`, even when embeddings are absent. Conditioning this check on `vector_length` preserves no-embedding ingestion on plain Valkey deployments.</violation>
</file>

_{Shadow auto-approve: would not auto-approve because issues were found.

Re-trigger cubic}

Make _index_exists gracefully return False when FT.LIST throws 'unknown command' (ValkeySearch not loaded), preserving text-only uploads on plain Valkey while keeping unconditional index detection for stale key cleanup. Signed-off-by: Anna Tao <annatao2004@gmail.com>

cubic-dev-ai

1 issue found across 1 file (changes from recent commits).

_{Shadow auto-approve: would not auto-approve because issues were found.
Tip: Review your code locally with the cubic CLI to iterate faster.

Re-trigger cubic}

Refactor _index_exists into _check_index returning (index_exists, search_module_available) tuple. When embeddings are present but the search module is missing, raise DestinationConnectionError before writing any hashes — preventing orphaned data. Preserves batch-then- create-index order (write before index creation) to avoid HNSW timeout. Signed-off-by: Anna Tao <annatao2004@gmail.com>

cubic-dev-ai

1 issue found across 1 file (changes from recent commits).

_{Shadow auto-approve: would not auto-approve because issues were found.
Tip: Review your code locally with the cubic CLI to iterate faster.

Re-trigger cubic}

Adds test_valkey_destination_vector_upload_fails_without_search_module that patches _check_index to return (False, False) and verifies DestinationConnectionError is raised before any HSET writes. Signed-off-by: Anna Tao <annatao2004@gmail.com>

cubic-dev-ai

0 issues found across 1 file (changes from recent commits).

_{Shadow auto-approve: would not auto-approve. Auto-approval blocked by 1 unresolved issue from previous reviews.

Re-trigger cubic}

Signed-off-by: Anna Tao <annatao2004@gmail.com>

atao2004 added 12 commits July 3, 2026 13:42

fix: add input validation for element_id and embeddings type

d5778eb

Refactor create_async_client to @asynccontextmanager for guaranteed c…

999f8d1

…leanup

timeout hardcoded fix, ssl=True changes

e4f1b96

helper extraction and refactor logic

aaae449

atao2004 requested a review from a team as a code owner July 3, 2026 20:45

cubic-dev-ai Bot reviewed Jul 3, 2026

View reviewed changes

cubic-dev-ai Bot reviewed Jul 6, 2026

View reviewed changes

Comment thread unstructured_ingest/processes/connectors/valkey.py Outdated

cubic-dev-ai Bot reviewed Jul 6, 2026

View reviewed changes

Comment thread unstructured_ingest/processes/connectors/valkey.py Outdated

cubic-dev-ai Bot reviewed Jul 6, 2026

View reviewed changes

Comment thread unstructured_ingest/processes/connectors/valkey.py

cubic-dev-ai Bot reviewed Jul 6, 2026

View reviewed changes

chore: trigger re-review

d2ded49

Signed-off-by: Anna Tao <annatao2004@gmail.com>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: add Valkey destination connector#747

feat: add Valkey destination connector#747
atao2004 wants to merge 17 commits into
Unstructured-IO:mainfrom
atao2004:integrate-valkey

atao2004 commented Jul 3, 2026 •

edited by cubic-dev-ai Bot

Loading

Uh oh!

socket-security Bot commented Jul 3, 2026 •

edited

Loading

Uh oh!

cubic-dev-ai Bot left a comment •

edited

Loading

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment •

edited

Loading

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment •

edited

Loading

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Uh oh!

Conversation

atao2004 commented Jul 3, 2026 • edited by cubic-dev-ai Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Why

Features

Testing

Dependencies

Changes

Uh oh!

socket-security Bot commented Jul 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

cubic-dev-ai Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

atao2004 commented Jul 3, 2026 •

edited by cubic-dev-ai Bot

Loading

socket-security Bot commented Jul 3, 2026 •

edited

Loading

cubic-dev-ai Bot left a comment •

edited

Loading

cubic-dev-ai Bot left a comment •

edited

Loading

cubic-dev-ai Bot left a comment •

edited

Loading