feat: added json extractor

api/server.py

@@ -27,6 +27,9 @@ System:
   GET  /docs          Swagger UI
   GET  /redoc         ReDoc UI
   GET  /metrics       Prometheus metrics
+
+Nested-JSON key extractor (under /api/v1/keys):
+  POST /api/v1/keys/extract
 """
 
 from __future__ import annotations
@@ -54,6 +57,7 @@ from pydantic import BaseModel, Field, field_validator
 
 from core import ConversionError, ConversionResult, DocumentConverter, SUPPORTED_EXTENSIONS
 from extraction.generic_json_extractor import extract
+from extraction.keys_extractor import Extractor as KeysExtractor
 from logger import get_logger
 from app.api.routes import router as pdf_router
 from app.core.auth import require_api_key
@@ -521,12 +525,108 @@ async def batch_urls(body: BatchUrlRequest):
     return BatchResponse(total=len(results), succeeded=succeeded, failed=len(results) - succeeded, total_time_ms=total_ms, results=results)
 
 
+# ---------------------------------------------------------------------------
+# Keys extractor router — nested-JSON key/value lookup
+# ---------------------------------------------------------------------------
+
+keys_router = APIRouter(dependencies=[Depends(require_api_key)])
+
+
+class KeysExtractRequest(BaseModel):
+    data: Any = Field(..., description="JSON object or array to search")
+    key_names: List[str] = Field(..., min_length=1, description="Key names to look up at any depth")
+    result_limit: Optional[int] = Field(
+        None,
+        ge=1,
+        description="Maximum values to return per key. Omit (or pass null) to return the full, uncapped result.",
+    )
+
+    @field_validator("key_names")
+    @classmethod
+    def _validate_key_names(cls, v: List[str]) -> List[str]:
+        for kn in v:
+            if not isinstance(kn, str) or not kn:
+                raise ValueError("each key_name must be a non-empty string")
+        return v
+
+
+class KeysExtractResponse(BaseModel):
+    success: bool
+    time_ms: float
+    data: Dict[str, List[Any]]
+    error_message: Optional[str] = None
+
+
+@keys_router.post(
+    "/extract",
+    response_model=KeysExtractResponse,
+    tags=["Keys Extractor"],
+    summary="Extract all values for given key names from a nested JSON object",
+)
+async def extract_keys(body: KeysExtractRequest):
+    """Recursively walk any JSON object/array and return every value attached
+    to the supplied key names, regardless of how deeply nested they are."""
+    start = time.perf_counter()
+    logger.info(
+        "keys_extract | start",
+        key_count=len(body.key_names),
+        limit=body.result_limit,
+    )
+
+    if not isinstance(body.data, (dict, list)):
+        return KeysExtractResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - start) * 1000, 3),
+            data={},
+            error_message="`data` must be a JSON object or array",
+        )
+
+    try:
+        loop = asyncio.get_running_loop()
+        results = await loop.run_in_executor(
+            _thread_pool,
+            KeysExtractor(body.data, body.key_names, body.result_limit).extract,
+        )
+    except (TypeError, ValueError) as exc:
+        logger.warning("keys_extract | invalid input", error=str(exc))
+        return KeysExtractResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - start) * 1000, 3),
+            data={},
+            error_message=str(exc),
+        )
+    except Exception as exc:
+        logger.exception("keys_extract | failed")
+        return KeysExtractResponse(
+            success=False,
+            time_ms=round((time.perf_counter() - start) * 1000, 3),
+            data={},
+            error_message=f"extraction failed: {exc}",
+        )
+
+    elapsed = round((time.perf_counter() - start) * 1000, 3)
+    total_values = sum(len(v) for v in results.values())
+    logger.info(
+        "keys_extract | done",
+        keys=len(body.key_names),
+        total_values=total_values,
+        time_ms=elapsed,
+    )
+    return KeysExtractResponse(
+        success=True,
+        time_ms=elapsed,
+        data=results,
+        error_message=None,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Include all routers
 # ---------------------------------------------------------------------------
 
 app.include_router(markitdown_router, prefix="/api/v1/markitdown")
 app.include_router(pdf_router, prefix="/api/v1")
+app.include_router(keys_router, prefix="/api/v1/keys")
 
 # -- Mount Prometheus metrics --
 app.mount("/metrics", make_asgi_app())

extraction/keys_extractor.py

@@ -0,0 +1,71 @@
+"""
+Nested JSON key-value extractor using jsonpath-ng.
+
+Walks any JSON-like structure (dict / list) and returns every value
+attached to a given key, regardless of how deeply nested it is.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from jsonpath_ng import parse
+from jsonpath_ng.exceptions import JsonPathLexerError, JsonPathParserError
+
+from logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class Extractor:
+    def __init__(
+        self,
+        data: dict | list,
+        key_names: list[str],
+        result_limit: int | None = 50,
+    ):
+        if not isinstance(key_names, list):
+            raise TypeError("key_names must be a list")
+        if not key_names:
+            raise ValueError("key_names must be a non-empty list")
+        for kn in key_names:
+            if not isinstance(kn, str) or not kn:
+                raise ValueError(f"each key_name must be a non-empty string, got {kn!r}")
+        if result_limit is not None and result_limit < 1:
+            raise ValueError("result_limit must be >= 1 or None")
+
+        self._data = data
+        self._key_names = key_names
+        self._result_limit = result_limit
+
+    def extract(self) -> dict[str, list[Any]]:
+        return {k: self._extract_single(k) for k in self._key_names}
+
+    def _extract_single(self, key_name: str) -> list[Any]:
+        try:
+            expr = _build_jp_expr(key_name)
+        except ValueError:
+            return []
+        vals = [m.value for m in expr.find(self._data)]
+        if self._result_limit is not None:
+            vals = vals[: self._result_limit]
+        return vals
+
+
+def _build_jp_expr(key_name: str) -> Any:
+    if '"' in key_name and "'" in key_name:
+        raise ValueError(f"key contains both quote types: {key_name!r}")
+    if '"' in key_name:
+        expr_str = "$..['" + key_name + "']"
+    else:
+        expr_str = '$..["' + key_name + '"]'
+    try:
+        return parse(expr_str)
+    except (JsonPathLexerError, JsonPathParserError):
+        raise ValueError(f"unable to build jsonpath expression for key: {key_name!r}")
+
+
+def get_nested_values(
+    data: dict | list,
+    key_name: str,
+) -> list[Any]:
+    return Extractor(data, [key_name], result_limit=None).extract()[key_name]

requirements.txt

@@ -20,6 +20,9 @@ pandas>=2.0.0
 # spaCy NER extraction for non-tabular file formats
 spacy>=3.7.0
 
+# Recursive nested-JSON key extraction
+jsonpath-ng>=1.8.0
+
 # CLI banner
 pyfiglet>=1.0.2
 rich>=13.0.0