Create a Buffer post with retry logic for rate limits. Retries on LimitReachedError responses from the GraphQL API.
(
access_token: str,
channel_id: str,
text: str,
media: Optional[dict] = None,
metadata: Optional[dict] = None,
scheduled_at: Optional[str] = None,
)
| 94 | |
| 95 | |
| 96 | def create_buffer_post_with_retry( |
| 97 | access_token: str, |
| 98 | channel_id: str, |
| 99 | text: str, |
| 100 | media: Optional[dict] = None, |
| 101 | metadata: Optional[dict] = None, |
| 102 | scheduled_at: Optional[str] = None, |
| 103 | ) -> dict: |
| 104 | """Create a Buffer post with retry logic for rate limits. |
| 105 | |
| 106 | Retries on LimitReachedError responses from the GraphQL API. |
| 107 | """ |
| 108 | for attempt in range(MAX_RETRIES): |
| 109 | result = create_buffer_post( |
| 110 | access_token=access_token, |
| 111 | channel_id=channel_id, |
| 112 | text=text, |
| 113 | media=media, |
| 114 | metadata=metadata, |
| 115 | scheduled_at=scheduled_at, |
| 116 | now=not scheduled_at, |
| 117 | ) |
| 118 | |
| 119 | if result.get("success"): |
| 120 | return result |
| 121 | |
| 122 | error = result.get("error", "") |
| 123 | # Retry on rate limit errors |
| 124 | if "limit" in error.lower() or "rate" in error.lower(): |
| 125 | if attempt < MAX_RETRIES - 1: |
| 126 | print(f"Rate limited, waiting {RETRY_DELAY}s before retry {attempt + 2}/{MAX_RETRIES}...") |
| 127 | time.sleep(RETRY_DELAY) |
| 128 | continue |
| 129 | |
| 130 | # Non-retryable error or max retries reached |
| 131 | return result |
| 132 | |
| 133 | return {"success": False, "error": "Max retries exceeded"} |
| 134 | |
| 135 | |
| 136 | def _instagram_metadata(post_data: dict, image_urls: list) -> dict: |
no test coverage detected