`services/search/service.py`

52 lines
"""SearchService — framework-agnostic anonymous-search core (stdlib).

Composes the anti-enumeration + privacy controls in ONE place so every transport (FastAPI,
a stdlib http.server, tests) enforces them identically:

    validate query  ->  rate-limit (per trusted client key)  ->  masked_search
                     ->  assert_public_safe on every row       ->  masked projection out

The authoritative property table is never queried directly by transports; they call this.
"""
from __future__ import annotations

from dataclasses import dataclass

from services.common.public_projection import assert_public_safe
from services.common.rate_limit import SlidingWindowRateLimiter


class EmptyQuery(ValueError):
    """Query had no alphanumeric token — rejected before hitting the index."""


class RateLimited(Exception):
    def __init__(self, retry_after: float) -> None:
        super().__init__(f"rate limited; retry after {retry_after:.1f}s")
        self.retry_after = retry_after


@dataclass
class SearchService:
    repository: object                       # provides masked_search(query, limit)
    limiter: SlidingWindowRateLimiter
    max_results: int = 20

    def search(self, query: str, client_key: str, limit: int | None = None) -> list[dict]:
        # 1) anti-enumeration: rate-limit BEFORE any work, keyed by trusted client id.
        if not self.limiter.allow(client_key):
            raise RateLimited(self.limiter.retry_after(client_key))

        # 2) reject empty/punctuation-only queries (mass-enumeration primitive).
        if not query or not query.strip():
            raise EmptyQuery("search query must contain at least one alphanumeric token")

        cap = min(limit or self.max_results, self.max_results)
        try:
            rows = self.repository.masked_search(query, limit=cap)
        except ValueError as exc:               # repository's own empty-query guard
            raise EmptyQuery(str(exc)) from exc

        # 3) fail closed: every row must be an allowlisted masked projection.
        return [assert_public_safe(r) for r in rows]