Setting the file. One moment. File Cu Canary
172helpers/_search_read.py
Python·172 lines·6 KB
except
ImportError
:
13 from _common import (
14 HelperFailure, HttpResult, ReadRecovery, Transport, odata_name, validate_search_endpoint,
15 )
16
17
18SUPPORTED_API_VERSIONS = {"2026-04-01", "2026-08-01-preview"}
19
20
21RESOURCE_SEGMENTS = {
22 "knowledge-source": "knowledgesources",
23 "knowledge-base": "knowledgebases",
24}
25
26
27DYNAMIC_FIELDS = {
28 "@odata.context",
29 "@odata.etag",
30 "currentSynchronizationState",
31 "lastSynchronizationState",
32 "synchronizationStatus",
33 "apiKey",
34 "createdResources",
35 # Azure Search always returns "<redacted>" for connectionString on GET
36 # (secret redaction), never the submitted value, so it can never be
37 # compared for exact equality against a desired definition.
38 "connectionString",
39}
40
41
42# Search normalizes these endpoint fields inconsistently on readback.
43URI_FIELDS = {"resourceUri", "uri"}
44
45
46SHA256 = re.compile(r"^sha256:[a-f0-9]{64}$")
47
48
49def _odata_name(name: Any) -> str:
50 return odata_name(name)
51
52
53def _resource_url(plan: dict[str, Any]) -> str:
54 endpoint = validate_search_endpoint(plan.get("endpoint"))
55 resource_type = plan.get("resource_type")
56 segment = RESOURCE_SEGMENTS.get(resource_type)
57 if segment is None:
58 raise HelperFailure(
59 "resource-type-invalid",
60 "resource_type must be knowledge-source or knowledge-base.",
61 blocked_at="input-resolution",
62 )
63 api_version = plan.get("api_version")
64 if api_version not in SUPPORTED_API_VERSIONS:
65 raise HelperFailure(
66 "api-version-invalid",
67 "API version must be 2026-04-01 or 2026-08-01-preview.",
68 blocked_at="input-resolution",
69 )
70 return (
71 f"{endpoint}/{segment}('{_odata_name(plan.get('name'))}')?"
72 + urlencode({"api-version": api_version})
73 )
74
75
76def _definition(value: Any, *, key: str | None = None) -> Any:
77 if isinstance(value, dict):
78 result = {
79 child_key: _definition(child, key=child_key)
80 for child_key, child in sorted(value.items())
81 if child_key not in DYNAMIC_FIELDS and child is not None
82 }
83 # Preview returns this documented default even when omitted on creation.
84 if value.get("kind") == "azureBlob" and result.get("resultsProcessing") == "rerank":
85 result.pop("resultsProcessing")
86 return {
87 child_key: child
88 for child_key, child in result.items()
89 if child not in ({}, [])
90 }
91 if isinstance(value, list):
92 return [_definition(child) for child in value]
93 # Azure Search silently strips a single trailing slash from
94 # azureOpenAIParameters.resourceUri on readback (while preserving it
95 # verbatim on aiServices.uri), so a byte-exact comparison would
96 # false-negative on functionally identical endpoints that differ only
97 # by a trailing slash. Normalize both known endpoint-URI field names.
98 if (
99 key in URI_FIELDS
100 and isinstance(value, str)
101 and value.endswith("/")
102 and len(value) > 1
103 ):
104 # Remove only one trailing slash; preserve intentional extra
105 # slashes (e.g. "https://example.com//") for exact comparison.
106 return value[:-1]
107 return value
108
109
110def response_etags(result: HttpResult) -> dict[str, Any]:
111 return {
112 "body": result.body.get("@odata.etag") if isinstance(result.body, dict) else None,
113 "headers": list(result.etag_values) if result.etag_values is not None else [
114 value for name, value in result.headers.items() if name.lower() == "etag"
115 ],
116 }
117
118
119def resolve_etag(evidence: dict[str, Any], request_id: str | None = None) -> str | None:
120 values = [*evidence["headers"]]
121 if evidence["body"] is not None:
122 values.append(evidence["body"])
123 if any(not isinstance(value, str) or not value.strip() for value in values):
124 raise HelperFailure("etag-invalid", "Search returned malformed version evidence.",
125 blocked_at="verification", request_id=request_id)
126 if len(set(values)) > 1:
127 raise HelperFailure("etag-conflict", "Search response header/body ETags conflict.",
128 blocked_at="verification", request_id=request_id)
129 return values[0] if values else None
130
131
132def _get(
133 url: str,
134 token: str,
135 *,
136 transport: Transport,
137 recovery: ReadRecovery | None = None,
138) -> tuple[dict[str, Any] | None, str | None]:
139 try:
140 result = (recovery.get(url, token, transport=transport) if recovery is not None
141 else transport("GET", url, token))
142 except HelperFailure as failure:
143 if failure.http_status == 404 and failure.blocked_at != "local-persistence":
144 return None, failure.request_id
145 raise
146 if result.status != 200 or not isinstance(result.body, dict):
147 raise HelperFailure(
148 "readback-invalid",
149 "Search resource readback did not return one JSON object.",
150 blocked_at="reconciliation",
151 request_id=result.request_id,
152 status=result.status,
153 )
154 etag = resolve_etag(response_etags(result), result.request_id)
155 body = {**result.body, "@odata.etag": etag} if etag is not None else result.body
156 return body, result.request_id
157
158
159def resource_url(plan: dict[str, Any]) -> str:
160 """Validate and address the exact selected Search resource."""
161 return _resource_url(plan)
162
163
164def read_resource(
165 url: str, token: str, *, transport: Transport
166) -> tuple[dict[str, Any] | None, str | None]:
167 """Read one identity; only a definitive 404 means absent."""
168 return _get(url, token, transport=transport)
169
170
171def definitions_match(desired: dict[str, Any], current: dict[str, Any]) -> bool:
172 return _definition(desired) == _definition(current)