JiriUhlir
232 lines
8.3 KiB
Python
232 lines
8.3 KiB
Python
"""HTTP client for the Meta Graph / Marketing API.
|
|
|
|
Deliberately hand-rolled over ``httpx`` rather than using the official
|
|
``facebook-business`` SDK: this service is a thin passthrough, the SDK is
|
|
synchronous and imposes its own object model and error types that we would only
|
|
have to translate back into our JSON shape. Staying on raw HTTP also means a new
|
|
Graph version is a config change, not a dependency bump.
|
|
|
|
What this module adds on top of a plain request:
|
|
|
|
* auth - the access token as a Bearer header (never a query param, so tokens
|
|
do not end up in upstream access logs), plus ``appsecret_proof`` when an app
|
|
secret was supplied;
|
|
* versioned URL building from the per-request Graph version;
|
|
* cursor pagination with an explicit page cap;
|
|
* error mapping into ``UpstreamError``;
|
|
* capture of Meta's rate-limit headers so callers can pace themselves.
|
|
|
|
Request and response bodies are otherwise forwarded as-is, so callers keep the
|
|
full upstream API surface.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from contextvars import ContextVar
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .. import config
|
|
from ..credentials import MetaCredentials
|
|
from ..errors import UpstreamError
|
|
from ..logging_config import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
# Meta reports quota consumption in these response headers. A middleware in
|
|
# main.py installs an empty dict per request and copies whatever lands in it
|
|
# onto our own response, so callers can see how close to a throttle they are
|
|
# without us threading a Response object through every route.
|
|
#
|
|
# The middleware must create the dict and we only ever MUTATE it: with
|
|
# Starlette's BaseHTTPMiddleware the endpoint runs in a child task that gets a
|
|
# *copy* of the context, so a `.set()` here would not be visible to the
|
|
# middleware - mutating the shared dict is.
|
|
USAGE_HEADERS = (
|
|
"X-App-Usage",
|
|
"X-Ad-Account-Usage",
|
|
"X-Business-Use-Case-Usage",
|
|
)
|
|
|
|
current_usage: ContextVar[dict[str, str] | None] = ContextVar(
|
|
"meta_usage", default=None
|
|
)
|
|
|
|
|
|
def _record_usage(resp: httpx.Response) -> None:
|
|
sink = current_usage.get()
|
|
if sink is None:
|
|
return
|
|
for header in USAGE_HEADERS:
|
|
if header in resp.headers:
|
|
sink[header] = resp.headers[header]
|
|
|
|
|
|
def _encode_form(data: dict[str, Any]) -> dict[str, str]:
|
|
"""Form-encode a body for Graph writes.
|
|
|
|
The Graph API expects form fields; anything structured (lists, dicts) has to
|
|
be a JSON string inside that form field, not a nested form structure.
|
|
"""
|
|
encoded: dict[str, str] = {}
|
|
for key, value in data.items():
|
|
if value is None:
|
|
continue
|
|
if isinstance(value, (dict, list)):
|
|
encoded[key] = json.dumps(value, ensure_ascii=False)
|
|
elif isinstance(value, bool):
|
|
encoded[key] = "true" if value else "false"
|
|
else:
|
|
encoded[key] = str(value)
|
|
return encoded
|
|
|
|
|
|
class MetaGraphClient:
|
|
"""Authenticated client for one Graph API request cycle."""
|
|
|
|
def __init__(
|
|
self,
|
|
creds: MetaCredentials,
|
|
*,
|
|
service_name: str = "Meta Graph API",
|
|
) -> None:
|
|
self._creds = creds
|
|
self._service_name = service_name
|
|
|
|
# --- URL / params ---------------------------------------------------------
|
|
def url(self, path: str) -> str:
|
|
"""Build a versioned Graph URL from a path like 'act_123/campaigns'."""
|
|
return f"{config.META_GRAPH_BASE_URL}/{self._creds.api_version}/{path.lstrip('/')}"
|
|
|
|
def build_url(self, url: str | httpx.URL, params: dict[str, Any] | None) -> httpx.URL:
|
|
"""Merge params and appsecret_proof INTO the URL's existing query.
|
|
|
|
Merging rather than passing httpx's ``params=`` matters: that argument
|
|
replaces the whole query string, which would silently strip the cursor
|
|
out of a ``paging.next`` URL and make pagination read page 1 forever.
|
|
"""
|
|
target = httpx.URL(url)
|
|
if params:
|
|
clean = {k: v for k, v in params.items() if v is not None}
|
|
if clean:
|
|
target = target.copy_merge_params(clean)
|
|
proof = self._creds.appsecret_proof()
|
|
if proof:
|
|
target = target.copy_merge_params({"appsecret_proof": proof})
|
|
return target
|
|
|
|
# --- requests -------------------------------------------------------------
|
|
async def request(
|
|
self,
|
|
method: str,
|
|
url: str | httpx.URL,
|
|
*,
|
|
params: dict[str, Any] | None = None,
|
|
form: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
headers = {"Authorization": f"Bearer {self._creds.access_token}"}
|
|
target = self.build_url(url, params)
|
|
try:
|
|
async with httpx.AsyncClient(timeout=config.HTTP_TIMEOUT_SECONDS) as client:
|
|
resp = await client.request(
|
|
method,
|
|
target,
|
|
data=_encode_form(form) if form is not None else None,
|
|
headers=headers,
|
|
)
|
|
except httpx.TimeoutException as exc:
|
|
raise UpstreamError(
|
|
f"{self._service_name} request timed out.", status=504
|
|
) from exc
|
|
except httpx.HTTPError as exc:
|
|
raise UpstreamError(
|
|
f"{self._service_name} is unreachable: {exc}", status=502
|
|
) from exc
|
|
|
|
_record_usage(resp)
|
|
return _parse_response(resp, self._service_name)
|
|
|
|
async def get(self, path: str, params: dict[str, Any] | None = None) -> Any:
|
|
return await self.request("GET", self.url(path), params=params)
|
|
|
|
async def post(
|
|
self,
|
|
path: str,
|
|
form: dict[str, Any] | None = None,
|
|
params: dict[str, Any] | None = None,
|
|
) -> Any:
|
|
return await self.request("POST", self.url(path), params=params, form=form or {})
|
|
|
|
async def get_absolute(self, url: str) -> Any:
|
|
"""GET an already-built absolute URL (used to follow paging.next)."""
|
|
return await self.request("GET", url)
|
|
|
|
# --- pagination -----------------------------------------------------------
|
|
async def get_all_pages(
|
|
self,
|
|
path: str,
|
|
params: dict[str, Any] | None = None,
|
|
*,
|
|
max_pages: int | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Follow ``paging.next`` and concatenate ``data`` across pages.
|
|
|
|
Stops at ``META_MAX_PAGES`` and marks the result ``truncated: true``
|
|
rather than looping unbounded or silently returning a partial list that
|
|
looks complete.
|
|
"""
|
|
cap = max_pages or config.META_MAX_PAGES
|
|
collected: list[Any] = []
|
|
page = await self.get(path, params)
|
|
pages_read = 1
|
|
|
|
while True:
|
|
if not isinstance(page, dict):
|
|
# Non-collection response - hand it back untouched.
|
|
return page
|
|
collected.extend(page.get("data") or [])
|
|
next_url = (page.get("paging") or {}).get("next")
|
|
if not next_url:
|
|
return {"data": collected, "pages_read": pages_read, "truncated": False}
|
|
if pages_read >= cap:
|
|
logger.warning(
|
|
"Paging cap reached for %s after %s pages; result truncated.",
|
|
path,
|
|
pages_read,
|
|
)
|
|
return {
|
|
"data": collected,
|
|
"pages_read": pages_read,
|
|
"truncated": True,
|
|
"next": next_url,
|
|
}
|
|
page = await self.get_absolute(next_url)
|
|
pages_read += 1
|
|
|
|
|
|
def _parse_response(resp: httpx.Response, service_name: str) -> Any:
|
|
try:
|
|
payload = resp.json()
|
|
except ValueError:
|
|
payload = {"raw": resp.text}
|
|
|
|
if resp.is_success:
|
|
return payload
|
|
|
|
# Graph errors look like {"error": {"message", "type", "code",
|
|
# "error_subcode", "error_user_title", "error_user_msg", "fbtrace_id"}}.
|
|
message = f"{service_name} error"
|
|
if isinstance(payload, dict):
|
|
err = payload.get("error")
|
|
if isinstance(err, dict):
|
|
# error_user_msg is the human-readable variant when Meta has one.
|
|
message = err.get("error_user_msg") or err.get("message") or message
|
|
|
|
raise UpstreamError(
|
|
message,
|
|
status=502 if resp.status_code >= 500 else resp.status_code,
|
|
upstream_status=resp.status_code,
|
|
body=payload,
|
|
)
|