| import os |
| import httpx |
| from typing import Any, Dict |
|
|
| class BackupClient: |
| _instance = None |
|
|
| def __new__(cls): |
| if cls._instance is None: |
| cls._instance = super(BackupClient, cls).__new__(cls) |
| cls._instance._initialized = False |
| return cls._instance |
|
|
| def __init__(self): |
| if self._initialized: |
| return |
| |
| self.backup_url = os.environ.get('BACKUP_URL') |
| if not self.backup_url: |
| raise ValueError("BACKUP_URL environment variable is not set") |
| |
| self._initialized = True |
| |
| async def push(self, data: Dict[str, Any]) -> Dict[str, Any]: |
| """ |
| Push data to Backup Sheet via POST request |
| |
| :param data: Dictionary containing the data to be pushed |
| :return: Response from the sheet |
| """ |
| try: |
| async with httpx.AsyncClient(follow_redirects=True) as client: |
| response = await client.post( |
| self.backup_url, |
| headers={ |
| "Content-Type": "application/json", |
| }, |
| json=data |
| ) |
| response.raise_for_status() |
| return response.json() |
|
|
| except httpx.HTTPStatusError as e: |
| raise Exception(f"HTTP error occurred: {e}") |
| except httpx.RequestError as e: |
| raise Exception(f"An error occurred while requesting: {e}") |
| except Exception as e: |
| raise Exception(f"An unexpected error occurred: {e}") |
