Execute a GraphQL request against the Buffer API.
(access_token: str, query: str, variables: Optional[Dict] = None)
| 60 | # --- Core helpers --- |
| 61 | |
| 62 | def _graphql_request(access_token: str, query: str, variables: Optional[Dict] = None) -> Dict: |
| 63 | """Execute a GraphQL request against the Buffer API.""" |
| 64 | headers = { |
| 65 | "Authorization": f"Bearer {access_token}", |
| 66 | "Content-Type": "application/json", |
| 67 | } |
| 68 | payload: Dict = {"query": query} |
| 69 | if variables: |
| 70 | payload["variables"] = variables |
| 71 | |
| 72 | response = requests.post( |
| 73 | BUFFER_GRAPHQL_URL, |
| 74 | headers=headers, |
| 75 | json=payload, |
| 76 | timeout=60, |
| 77 | ) |
| 78 | |
| 79 | if response.status_code != 200: |
| 80 | print(f"Buffer GraphQL HTTP error: {response.status_code} - {response.text[:500]}") |
| 81 | return {"errors": [{"message": f"HTTP {response.status_code}"}]} |
| 82 | |
| 83 | result = response.json() |
| 84 | if "errors" in result: |
| 85 | print(f"Buffer GraphQL errors: {result['errors']}") |
| 86 | return result |
| 87 | |
| 88 | |
| 89 | def get_organization_id(access_token: str) -> Optional[str]: |
no test coverage detected