Source Readback

1"""Readback guard shared by source creation and retained-source verification."""
2from __future__ import annotations
3
4from typing import Any
5
6try:
7    from ._common import HelperFailure, Transport
8    from . import _embedding, _search_read
9except ImportError:
10    from _common import HelperFailure, Transport
11    import _embedding
12    import _search_read
13
14
15def guard_readback_transport(plan: dict[str, Any], transport: Transport) -> Transport:
16    if "embedding" not in plan and "content_understanding" not in plan:
17        return transport
18    url = _search_read.resource_url(plan["source"])
19
20    def guarded(method: str, target: str, token: str, **kwargs: Any):
21        response = transport(method, target, token, **kwargs)
22        if method == "GET" and target == url and response.status == 200:
23            try:
24                _embedding.verify_source_readback(plan.get("embedding"), response.body)
25                if "content_understanding" in plan:
26                    try:
27                        from . import _blob_source_read as blob_source, _file_source_config as file_source
28                    except ImportError:
29                        import _blob_source_read as blob_source, _file_source_config as file_source
30                    owner = file_source if plan["source"]["desired"]["kind"] == "file" else blob_source
31                    owner.verify_content_understanding_readback(plan["content_understanding"], response.body)
32            except HelperFailure as failure:
33                failure.request_id = response.request_id
34                raise
35        return response
36
37    return guarded