mirror of
https://github.com/cocoindex-io/cocoindex-code.git
synced 2026-09-14 16:39:38 +08:00
Vasilii Alferov
f2ecfb5df3
* chore: enable mypy explicit_package_bases, remove stale type: ignore comments explicit_package_bases = true resolves module paths from the repo root, preventing double-discovery of files in tests/ under different module names. Required for tests/example_toml_chunker.py to be importable as example_toml_chunker rather than an ambiguous bare module. Also sets asyncio_mode = auto so pytest-asyncio behaviour is explicit (default in 1.3.0 is strict). With explicit_package_bases active, mypy can fully resolve the Response union in test_daemon.py — the 16 type: ignore[union-attr] and type: ignore[attr-defined] comments that were suppressing false positives under the old resolution are now unused and removed. # Conflicts: # tests/test_daemon.py * feat: pluggable chunker registry with settings integration Improves retrieval precision by letting users split specific file types at semantic boundaries (e.g. TOML sections, SQL statements) instead of the default line-window splitter. ## What - cocoindex_code/chunking.py: public API module exporting ChunkerFn (Callable alias), CHUNKER_REGISTRY context key, and re-exports of Chunk/TextPosition from upstream. Single import path for chunker authors. - tests/example_toml_chunker.py: demo chunker splitting at [section] headers; excludes [[array_of_tables]] via negative lookahead. Lives in tests/ to signal it belongs to a separate package, not the core library. - ProjectSettings.chunkers: new list[ChunkerMapping] field, serialised as YAML. Each entry maps a file extension to a 'module.path:callable' string. Users activate chunkers by editing .cocoindex_code/settings.yml — no code changes required. - daemon.py: _resolve_chunker_registry resolves ChunkerMapping entries via importlib at project load time and passes the result to Project.create(). callable() guard gives a clear error at startup rather than a TypeError per file. - Project.create(chunker_registry=...): new optional parameter. Injected as a cocoindex context key (tracked=False) rather than exposed via env internals. Empty registry by default — zero behavioural delta for existing users. - indexer.py: process_file checks the registry per file suffix; falls through to RecursiveSplitter unchanged when no chunker is registered. ## Design decisions - ChunkerFn returns (language_override, chunks): language_override=None keeps detect_code_language() result; non-None lets the chunker correct it (e.g. .sls files starting with #!py). - tracked=False is consistent with SQLITE_DB, CODEBASE_DIR, and other non-serialisable context keys. Changing a chunker requires a daemon restart, which triggers a full re-index anyway. - _resolve_chunker_registry lives in daemon.py, its only call site, keeping settings.py as pure schema/IO and chunking.py as pure type definitions. # Conflicts: # src/cocoindex_code/daemon.py # src/cocoindex_code/indexer.py # src/cocoindex_code/project.py # tests/test_settings.py
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
"""Demo chunker: splits TOML files at top-level [section] boundaries.
|
|
|
|
Each ``[section]`` header starts a new chunk, keeping the section header
|
|
and its key-value pairs together. This produces semantically coherent units
|
|
instead of the arbitrary line-window slices from the default splitter.
|
|
|
|
Register in ``.cocoindex_code/settings.yml``::
|
|
|
|
chunkers:
|
|
- ext: toml
|
|
module: example_toml_chunker:toml_chunker
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re as _re
|
|
from pathlib import Path as _Path
|
|
|
|
from cocoindex_code.chunking import Chunk, TextPosition
|
|
|
|
_SECTION_RE = _re.compile(r"^\[(?!\[)")
|
|
|
|
|
|
def _pos(line: int) -> TextPosition:
|
|
return TextPosition(byte_offset=0, char_offset=0, line=line, column=0)
|
|
|
|
|
|
def toml_chunker(path: _Path, content: str) -> tuple[str | None, list[Chunk]]:
|
|
"""Split a TOML file at top-level ``[section]`` headers."""
|
|
lines = content.splitlines()
|
|
section_starts = [i for i, ln in enumerate(lines) if _SECTION_RE.match(ln)]
|
|
|
|
if not section_starts:
|
|
return "toml", [Chunk(text=content, start=_pos(1), end=_pos(len(lines)))]
|
|
|
|
boundaries = section_starts + [len(lines)]
|
|
chunks: list[Chunk] = []
|
|
for start_idx, end_idx in zip(boundaries, boundaries[1:]):
|
|
text = "\n".join(lines[start_idx:end_idx]).strip()
|
|
if text:
|
|
chunks.append(Chunk(text=text, start=_pos(start_idx + 1), end=_pos(end_idx)))
|
|
return "toml", chunks
|
|
|
|
|
|
__all__ = ["toml_chunker"]
|