Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
75 changes: 1 addition & 74 deletions db.py
Original file line number Diff line number Diff line change
Expand Up @@ -349,7 +349,7 @@ def transaction(db: sqlite3.Connection):
Example:
with transaction(db):
for item in items:
upsert_symbol(db, ..., auto_commit=False)
upsert_file(db, ..., auto_commit=False)
# Single commit here
"""
# Disable autocommit by starting a transaction
Expand Down Expand Up @@ -672,79 +672,6 @@ def delete_file_data(db: sqlite3.Connection, file_id: int, auto_commit: bool = T
db.commit()


def upsert_symbol(
db: sqlite3.Connection,
name: str,
kind: str,
file_id: int,
line_start: int,
line_end: int,
parent_symbol_id: int | None,
source_text: str,
auto_commit: bool = True,
) -> int:
"""Insert or update a symbol record. Returns the symbol_id."""
db.execute(
"""
INSERT INTO symbols (name, kind, file_id, line_start, line_end,
parent_symbol_id, source_text)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(file_id, name, kind, line_start) DO UPDATE SET
line_end = excluded.line_end,
parent_symbol_id = excluded.parent_symbol_id,
source_text = excluded.source_text
""",
(name, kind, file_id, line_start, line_end, parent_symbol_id, source_text),
)
if auto_commit:
db.commit()
row = db.execute(
"SELECT id FROM symbols WHERE file_id = ? AND name = ? AND kind = ? AND line_start = ?",
(file_id, name, kind, line_start),
).fetchone()
return row[0]


def upsert_reference(
db: sqlite3.Connection,
symbol_name: str,
file_id: int,
line_number: int,
auto_commit: bool = True,
) -> None:
"""Insert or update a cross-reference record."""
db.execute(
"""
INSERT INTO references_ (symbol_name, file_id, line_number)
VALUES (?, ?, ?)
ON CONFLICT(symbol_name, file_id, line_number) DO NOTHING
""",
(symbol_name, file_id, line_number),
)
if auto_commit:
db.commit()


def upsert_embedding(
db: sqlite3.Connection,
symbol_id: int,
embedding: list[float],
auto_commit: bool = True,
) -> None:
"""Insert or replace a symbol's dense vector embedding."""
import struct

blob = struct.pack(f"{len(embedding)}f", *embedding)
# sqlite-vec doesn't support ON CONFLICT, so delete-then-insert
db.execute("DELETE FROM symbol_embeddings WHERE symbol_id = ?", (symbol_id,))
db.execute(
"INSERT INTO symbol_embeddings (symbol_id, embedding) VALUES (?, ?)",
(symbol_id, blob),
)
if auto_commit:
db.commit()


def batch_insert_embeddings(
db: sqlite3.Connection,
pairs: list[tuple[int, list[float]]],
Expand Down
10 changes: 0 additions & 10 deletions logging_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,16 +199,6 @@ def get_db_logger() -> logging.Logger:
return get_logger("db")


def get_parser_logger() -> logging.Logger:
"""Get logger for parser module."""
return get_logger("parser")


def get_query_logger() -> logging.Logger:
"""Get logger for query module."""
return get_logger("queries")


def get_git_logger() -> logging.Logger:
"""Get logger for git search module."""
return get_logger("git")
33 changes: 0 additions & 33 deletions validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,39 +232,6 @@ def validate_top_k(value: int, min_val: int = 1, max_val: int = 100, default: in
return value


def validate_path_in_directory(path: str, base_dir: str) -> Path:
"""Validate that a path is within a base directory (prevent path traversal).

Args:
path: Path to validate
base_dir: Base directory that path must be within

Returns:
Resolved Path object

Raises:
ValidationError: If path escapes base directory
"""
if not path:
raise ValidationError("Path cannot be empty")

try:
resolved_path = Path(path).resolve()
resolved_base = Path(base_dir).resolve()
except Exception as e:
raise ValidationError(f"Invalid path: {path}", {"exception": str(e)})

try:
resolved_path.relative_to(resolved_base)
except ValueError:
raise ValidationError(
f"Path escapes base directory: {path}",
{"base_directory": str(resolved_base)}
)

return resolved_path


def sanitize_fts_query(query: str) -> str:
"""Sanitize a query for FTS5 MATCH with good recall.

Expand Down
Loading