Source code for sf_toolkit.resources.metadata_soap
"""Schema-independent Metadata SOAP read/update operations.
Records retain their XML tree so nested and unknown fields survive an update.
Session headers and OAuth metadata can contain secrets: repr/errors never dump
XML, fault messages, or record values. Access those values explicitly instead.
"""
from dataclasses import dataclass
import re
from lxml import etree
from ..auth import SalesforceAuth
from ..client import AsyncSalesforceClient, SalesforceClient
from ..request_policy import NO_REPLAY
from .base import ApiResource
SOAP = "http://schemas.xmlsoap.org/soap/envelope/"
META = "http://soap.sforce.com/2006/04/metadata"
XSI = "http://www.w3.org/2001/XMLSchema-instance"
_TYPE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
[docs]
class MetadataResponseError(ValueError):
"""An invalid/unsuccessful SOAP response; raw payloads are not retained."""
[docs]
class MetadataSoapFault(Exception):
"""SOAP fault values are available explicitly, but excluded from str/repr."""
[docs]
def __init__(self, code: str, message: str, status_code: int):
self.code = code
self.fault_message = message
self.status_code = status_code
super().__init__("Salesforce Metadata SOAP fault")
def _element(parent, name: str, text: str | None = None):
node = etree.SubElement(parent, f"{{{META}}}{name}")
node.text = text
return node
def _metadata_type(value: str) -> str:
if not isinstance(value, str) or not _TYPE.fullmatch(value):
raise ValueError("Invalid Metadata API type name")
return value
def _xml_parser(*, clean=False):
return etree.XMLParser(
resolve_entities=False,
no_network=True,
load_dtd=False,
remove_blank_text=clean,
remove_comments=clean,
)
def _clone_xml(node):
# deepcopy drops inherited aliases used only in QName *values*. Serializing
# in the original context materializes every in-scope namespace, including
# nested prefix shadowing. Never reparent the clone into another XML tree.
return etree.fromstring(etree.tostring(node, with_tail=False), _xml_parser())
[docs]
@dataclass(repr=False)
class MetadataRecord:
"""A Metadata XML tree. Edit scalars with get/set or nested XML via element.
Existing child order, namespaces, attributes and unknown fields are retained.
New fields must be placed in the order required by the Salesforce schema.
"""
metadata_type: str
element: etree._Element
def __post_init__(self):
_metadata_type(self.metadata_type)
[docs]
def get(self, name: str) -> str | None:
_metadata_type(name)
nodes = self.element.findall(f"{{{META}}}{name}")
if not nodes:
return None
if len(nodes) != 1 or len(nodes[0]):
raise ValueError("Metadata field is not a single scalar")
if nodes[0].get(f"{{{XSI}}}nil") in {"true", "1"}:
return None
return nodes[0].text or ""
[docs]
def set(self, name: str, value: str | None) -> None:
_metadata_type(name)
nodes = self.element.findall(f"{{{META}}}{name}")
if len(nodes) > 1 or (nodes and len(nodes[0])):
raise ValueError("Metadata field is not a single scalar")
node = nodes[0] if nodes else _element(self.element, name)
node.text = value
if value is None:
node.set(f"{{{XSI}}}nil", "true")
else:
node.attrib.pop(f"{{{XSI}}}nil", None)
@property
def full_name(self) -> str | None:
return self.get("fullName")
[docs]
def copy(self) -> "MetadataRecord":
return type(self)(self.metadata_type, _clone_xml(self.element))
[docs]
@dataclass(frozen=True, repr=False)
class MetadataError:
status_code: str
message: str
fields: tuple[str, ...] = ()
[docs]
@dataclass(frozen=True)
class MetadataSaveResult:
full_name: str
success: bool
errors: tuple[MetadataError, ...] = ()
class _MetadataSoap:
# Bounds our response buffer, not upstream transport allocations or the XML
# tree. Compressed responses are rejected before consuming an unread stream.
max_response_bytes: int = 10 * 1024 * 1024
client: SalesforceClient | AsyncSalesforceClient
def _auth(self) -> SalesforceAuth:
auth = self.client.auth
if not isinstance(auth, SalesforceAuth):
raise ValueError("Metadata SOAP requires SalesforceAuth")
return auth
@staticmethod
def _batch(metadata_type: str, values: tuple) -> None:
_metadata_type(metadata_type)
limit = 200 if metadata_type in {"CustomMetadata", "CustomApplication"} else 10
if not 1 <= len(values) <= limit:
raise ValueError(f"Metadata call requires 1 to {limit} components")
def _read_content(self, metadata_type: str, full_names: tuple[str, ...]):
self._batch(metadata_type, full_names)
if any(not isinstance(name, str) or not name.strip() for name in full_names):
raise ValueError("Metadata full names must be nonempty strings")
if len(set(full_names)) != len(full_names):
raise ValueError("Duplicate metadata full names")
content = etree.Element("content")
_element(content, "type", metadata_type)
for name in full_names:
_element(content, "fullNames", name)
return list(content)
def _update_content(self, records: tuple[MetadataRecord, ...]):
if not records or any(
not isinstance(record, MetadataRecord) for record in records
):
raise ValueError("Metadata update requires MetadataRecord instances")
metadata_type = records[0].metadata_type
self._batch(metadata_type, records)
names = []
content = []
for record in records:
if record.metadata_type != metadata_type:
raise ValueError("Metadata update must contain only one component type")
if not record.full_name or not record.full_name.strip():
raise ValueError("Metadata update requires fullName")
names.append(record.full_name)
node = _clone_xml(record.element)
node.tag = f"{{{META}}}metadata"
# QName-valued node.set can segfault in supported lxml/libxml2 when
# META is the default namespace. Bind the QName as ordinary text;
# an unprefixed QName resolves against the default namespace.
prefix = next(key for key, uri in node.nsmap.items() if uri == META)
declared_type = f"{prefix}:{metadata_type}" if prefix else metadata_type
node.set(f"{{{XSI}}}type", declared_type)
content.append(node)
if len(set(names)) != len(names):
raise ValueError("Duplicate metadata full names")
return content
def _request(self, operation: str, content):
auth = self._auth()
token = auth.token
if token is None:
raise ValueError("Metadata SOAP requires an authenticated session")
# A concurrent REST refresh may change auth.token while we serialize.
# Bind the endpoint, origin check and SOAP credential to one session.
version = self.client.data_url.rsplit("/", 1)[-1].removeprefix("v")
url = token.instance.join(f"/services/Soap/m/{version}")
auth._check_resource_url(url, token)
header = etree.Element(
f"{{{SOAP}}}Header", nsmap={"soapenv": SOAP, "met": META}
)
_element(_element(header, "SessionHeader"), "sessionId", token.token)
# Serialize fragments independently. Reparenting lxml nodes can discard
# namespace aliases referenced only by QName-valued attributes (xsi:type),
# silently corrupting unknown/nested metadata types on a round trip.
fragments = b"".join(
etree.tostring(child, with_tail=False) for child in content
)
data = (
b'<?xml version="1.0" encoding="UTF-8"?>'
+ f'<soapenv:Envelope xmlns:soapenv="{SOAP}" xmlns:met="{META}">'.encode()
+ etree.tostring(header)
+ f"<soapenv:Body><met:{operation}>".encode()
+ fragments
+ f"</met:{operation}></soapenv:Body></soapenv:Envelope>".encode()
)
return url, data
@staticmethod
def _parse(operation: str, status: int, data: bytes):
try:
root = etree.fromstring(data, _xml_parser(clean=True))
if root.getroottree().docinfo.doctype:
raise MetadataResponseError(
"DOCTYPE is not permitted in Metadata responses"
)
except etree.LxmlError:
raise MetadataResponseError("Invalid Metadata SOAP XML response") from None
if root.tag != f"{{{SOAP}}}Envelope":
raise MetadataResponseError("Invalid Metadata SOAP envelope")
bodies = root.findall(f"{{{SOAP}}}Body")
if len(bodies) != 1 or len(bodies[0]) != 1:
raise MetadataResponseError("Invalid Metadata SOAP body")
result = bodies[0][0]
if result.tag == f"{{{SOAP}}}Fault":
raise MetadataSoapFault(
result.findtext("faultcode", ""),
result.findtext("faultstring", ""),
status,
)
if status != 200:
raise MetadataResponseError(f"Metadata SOAP request failed (HTTP {status})")
if result.tag != f"{{{META}}}{operation}Response":
raise MetadataResponseError("Unexpected Metadata SOAP operation response")
return result
@staticmethod
def _records(response, metadata_type: str, full_names: tuple[str, ...]):
results = response.findall(f"{{{META}}}result")
if len(results) != 1:
raise MetadataResponseError("Invalid Metadata read result")
records = []
found = set()
for node in results[0].findall(f"{{{META}}}records"):
if node.get(f"{{{XSI}}}nil") in {"true", "1"}:
continue
declared_type = node.get(f"{{{XSI}}}type")
if declared_type:
prefix, _, local_type = declared_type.rpartition(":")
if (
local_type != metadata_type
or node.nsmap.get(prefix or None) != META
):
raise MetadataResponseError("Unexpected Metadata record type")
record = MetadataRecord(metadata_type, _clone_xml(node))
try:
full_name = record.full_name
except ValueError:
raise MetadataResponseError(
"Invalid Metadata record fullName"
) from None
if full_name not in full_names or full_name in found:
raise MetadataResponseError("Unexpected Metadata record fullName")
found.add(full_name)
records.append(record)
return records
@staticmethod
def _results(response, records: tuple[MetadataRecord, ...]):
results = []
expected = {record.full_name for record in records}
found = set()
for node in response.findall(f"{{{META}}}result"):
record = MetadataRecord(records[0].metadata_type, node)
try:
name, success = record.full_name, record.get("success")
except ValueError:
raise MetadataResponseError("Invalid Metadata save result") from None
if (
name not in expected
or name in found
or success not in {"true", "false", "1", "0"}
):
raise MetadataResponseError("Invalid Metadata save result")
found.add(name)
errors = tuple(
MetadataError(
error.findtext(f"{{{META}}}statusCode", ""),
error.findtext(f"{{{META}}}message", ""),
tuple(error.itertext(f"{{{META}}}fields")),
)
for error in node.findall(f"{{{META}}}errors")
)
if errors and success in {"true", "1"}:
raise MetadataResponseError("Contradictory Metadata save result")
results.append(MetadataSaveResult(name, success in {"true", "1"}, errors))
if found != expected:
raise MetadataResponseError("Missing Metadata save results")
return results
@staticmethod
def _check_encoding(response):
encoding = response.headers.get("Content-Encoding")
if encoding is not None and any(
part.strip().lower() != "identity" for part in encoding.split(",")
):
raise MetadataResponseError(
"Compressed/encoded Metadata SOAP responses are not supported"
)
def _append_response(self, data: bytearray, chunk: bytes):
if len(chunk) > self.max_response_bytes - len(data):
raise MetadataResponseError("Metadata SOAP response exceeds size limit")
data.extend(chunk)
[docs]
class MetadataSoapResource(_MetadataSoap, ApiResource):
client: SalesforceClient
def _session(self):
if self._auth().token is None:
self.client.get("/services/oauth2/userinfo")
if self.client.api_version is None:
versions = self.client.versions
self.client.api_version = versions[max(versions)]
def _call(self, operation: str, content):
self._session()
url, data = self._request(operation, content)
with self.client.stream(
"POST",
url,
content=data,
auth=None,
follow_redirects=False,
extensions=NO_REPLAY.extensions,
headers={
"Content-Type": "text/xml; charset=UTF-8",
"SOAPAction": '""',
"Accept-Encoding": "identity",
},
) as response:
self._check_encoding(response)
data = bytearray()
# Custom transports/hooks may already have buffered/decoded content;
# we can bound our own copy, but cannot undo those allocations.
if response.is_stream_consumed:
self._append_response(data, response.content)
else:
for chunk in response.iter_raw():
self._append_response(data, chunk)
return self._parse(operation, response.status_code, bytes(data))
[docs]
def read(self, metadata_type: str, *full_names: str) -> list[MetadataRecord]:
"""Read up to 10 components (200 for CustomMetadata/CustomApplication).
Returns found records; nil/missing components are omitted. No automatic
retries or auth replay occur; an expired SOAP session raises a fault.
"""
content = self._read_content(metadata_type, full_names)
return self._records(
self._call("readMetadata", content), metadata_type, full_names
)
[docs]
def update(self, *records: MetadataRecord) -> list[MetadataSaveResult]:
"""Submit one update, retaining every field in each record's XML tree.
Inspect every returned success/errors. Transport failures have an unknown
outcome: re-read/reconcile rather than blindly resubmitting this call.
"""
content = self._update_content(records)
return self._results(self._call("updateMetadata", content), records)
[docs]
class AsyncMetadataSoapResource(_MetadataSoap):
client: AsyncSalesforceClient
async def _session(self):
if self._auth().token is None:
await self.client.get("/services/oauth2/userinfo")
if self.client.api_version is None:
versions = await self.client.versions()
self.client.api_version = versions[max(versions)]
async def _call(self, operation: str, content):
await self._session()
url, data = self._request(operation, content)
async with self.client.stream(
"POST",
url,
content=data,
auth=None,
follow_redirects=False,
extensions=NO_REPLAY.extensions,
headers={
"Content-Type": "text/xml; charset=UTF-8",
"SOAPAction": '""',
"Accept-Encoding": "identity",
},
) as response:
self._check_encoding(response)
data = bytearray()
if response.is_stream_consumed:
self._append_response(data, response.content)
else:
async for chunk in response.aiter_raw():
self._append_response(data, chunk)
return self._parse(operation, response.status_code, bytes(data))
[docs]
async def read(self, metadata_type: str, *full_names: str) -> list[MetadataRecord]:
content = self._read_content(metadata_type, full_names)
response = await self._call("readMetadata", content)
return self._records(response, metadata_type, full_names)
[docs]
async def update(self, *records: MetadataRecord) -> list[MetadataSaveResult]:
content = self._update_content(records)
return self._results(await self._call("updateMetadata", content), records)