Maxim Snesarev
33d6bb7ebd
- Updated database configuration to switch from SQLite to PostgreSQL, including changes to alembic.ini, Docker Compose, and environment settings. - Refactored application code to utilize PostgreSQL database adapters, ensuring compatibility with the new database structure. - Enhanced API routes and data handling to support the new database, including adjustments in data models and query logic. - Introduced new job processing mechanisms for full synchronization of AMO CRM entities, leveraging FastStream for background tasks. - Improved logging and error handling across the application to facilitate better monitoring and debugging. - Removed obsolete SQLite adapter files and migrations, streamlining the project structure for PostgreSQL integration.
370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""
|
|
Google Sheets API Client for exporting AMO CRM data.
|
|
|
|
This adapter handles authentication and data export to Google Sheets
|
|
using service account credentials.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from typing import List, Any, Optional
|
|
from google.oauth2 import service_account
|
|
from googleapiclient.discovery import build
|
|
from googleapiclient.errors import HttpError
|
|
from utils.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GoogleSheetsClient:
|
|
"""Client for interacting with Google Sheets API."""
|
|
|
|
def __init__(self, service_account_file: Optional[str] = None):
|
|
"""
|
|
Initialize Google Sheets client with service account credentials.
|
|
|
|
Args:
|
|
service_account_file: Path to service account JSON file
|
|
"""
|
|
self.service_account_file = service_account_file or settings.GOOGLE_SERVICE_ACCOUNT_FILE
|
|
self.scopes = [settings.GOOGLE_SCOPES]
|
|
self.service = None
|
|
|
|
if self.service_account_file:
|
|
self._authenticate()
|
|
|
|
def _authenticate(self) -> None:
|
|
"""Authenticate with Google Sheets API using service account."""
|
|
try:
|
|
credentials = service_account.Credentials.from_service_account_file(
|
|
self.service_account_file,
|
|
scopes=self.scopes
|
|
)
|
|
|
|
self.service = build('sheets', 'v4', credentials=credentials)
|
|
logger.info("Successfully authenticated with Google Sheets API")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to authenticate with Google Sheets API: {str(e)}")
|
|
raise
|
|
|
|
async def write_data(
|
|
self,
|
|
spreadsheet_id: str,
|
|
sheet_name: str,
|
|
data: List[List[Any]],
|
|
clear_existing: bool = True
|
|
) -> dict:
|
|
"""
|
|
Write data to a Google Sheet.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab to write to
|
|
data: 2D array of data to write (rows x columns)
|
|
clear_existing: Whether to clear existing data before writing
|
|
|
|
Returns:
|
|
Dictionary with update result information
|
|
|
|
Raises:
|
|
HttpError: If the API request fails
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
# Ensure sheet exists, create if not
|
|
await self._ensure_sheet_exists(spreadsheet_id, sheet_name)
|
|
|
|
# Clear existing data if requested
|
|
if clear_existing:
|
|
await self.clear_sheet(spreadsheet_id, sheet_name)
|
|
|
|
# Prepare the update request
|
|
range_name = f"{sheet_name}!A1"
|
|
body = {
|
|
'values': data,
|
|
'majorDimension': 'ROWS'
|
|
}
|
|
|
|
# Execute the update
|
|
result = self.service.spreadsheets().values().update(
|
|
spreadsheetId=spreadsheet_id,
|
|
range=range_name,
|
|
valueInputOption='USER_ENTERED', # Parse formulas and format numbers
|
|
body=body
|
|
).execute()
|
|
|
|
updated_cells = result.get('updatedCells', 0)
|
|
logger.info(
|
|
f"Successfully wrote {len(data)} rows to sheet '{sheet_name}' "
|
|
f"({updated_cells} cells updated)"
|
|
)
|
|
|
|
return {
|
|
'updated_rows': len(data),
|
|
'updated_cells': updated_cells,
|
|
'updated_range': result.get('updatedRange')
|
|
}
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to write data to Google Sheets: {str(e)}")
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error writing to Google Sheets: {str(e)}")
|
|
raise
|
|
|
|
async def clear_sheet(self, spreadsheet_id: str, sheet_name: str) -> dict:
|
|
"""
|
|
Clear all data from a sheet.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab to clear
|
|
|
|
Returns:
|
|
Dictionary with clear result information
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
range_name = f"{sheet_name}!A:Z" # Clear columns A through Z
|
|
|
|
result = self.service.spreadsheets().values().clear(
|
|
spreadsheetId=spreadsheet_id,
|
|
range=range_name,
|
|
body={}
|
|
).execute()
|
|
|
|
logger.info(f"Successfully cleared sheet '{sheet_name}'")
|
|
return result
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to clear sheet: {str(e)}")
|
|
raise
|
|
|
|
async def _ensure_sheet_exists(self, spreadsheet_id: str, sheet_name: str) -> None:
|
|
"""
|
|
Ensure a sheet with the given name exists, create it if not.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab to check/create
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
# Get existing sheets
|
|
spreadsheet = self.service.spreadsheets().get(
|
|
spreadsheetId=spreadsheet_id
|
|
).execute()
|
|
|
|
sheets = spreadsheet.get('sheets', [])
|
|
sheet_names = [sheet['properties']['title'] for sheet in sheets]
|
|
|
|
# Check if sheet exists
|
|
if sheet_name in sheet_names:
|
|
logger.debug(f"Sheet '{sheet_name}' already exists")
|
|
return
|
|
|
|
# Create new sheet
|
|
logger.info(f"Creating new sheet '{sheet_name}'")
|
|
|
|
request = {
|
|
'requests': [{
|
|
'addSheet': {
|
|
'properties': {
|
|
'title': sheet_name
|
|
}
|
|
}
|
|
}]
|
|
}
|
|
|
|
self.service.spreadsheets().batchUpdate(
|
|
spreadsheetId=spreadsheet_id,
|
|
body=request
|
|
).execute()
|
|
|
|
logger.info(f"Successfully created sheet '{sheet_name}'")
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to ensure sheet exists: {str(e)}")
|
|
raise
|
|
|
|
async def append_data(
|
|
self,
|
|
spreadsheet_id: str,
|
|
sheet_name: str,
|
|
data: List[List[Any]]
|
|
) -> dict:
|
|
"""
|
|
Append data to the end of a sheet.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab to append to
|
|
data: 2D array of data to append
|
|
|
|
Returns:
|
|
Dictionary with append result information
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
# Ensure sheet exists
|
|
await self._ensure_sheet_exists(spreadsheet_id, sheet_name)
|
|
|
|
range_name = f"{sheet_name}!A:A" # Append starting at column A
|
|
body = {
|
|
'values': data,
|
|
'majorDimension': 'ROWS'
|
|
}
|
|
|
|
result = self.service.spreadsheets().values().append(
|
|
spreadsheetId=spreadsheet_id,
|
|
range=range_name,
|
|
valueInputOption='USER_ENTERED',
|
|
insertDataOption='INSERT_ROWS',
|
|
body=body
|
|
).execute()
|
|
|
|
logger.info(f"Successfully appended {len(data)} rows to sheet '{sheet_name}'")
|
|
return result
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to append data to Google Sheets: {str(e)}")
|
|
raise
|
|
|
|
async def batch_update(
|
|
self,
|
|
spreadsheet_id: str,
|
|
updates: List[dict]
|
|
) -> dict:
|
|
"""
|
|
Perform batch updates to multiple sheets.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
updates: List of update operations
|
|
|
|
Returns:
|
|
Dictionary with batch update results
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
result = self.service.spreadsheets().batchUpdate(
|
|
spreadsheetId=spreadsheet_id,
|
|
body={'requests': updates}
|
|
).execute()
|
|
|
|
logger.info(f"Successfully performed {len(updates)} batch updates")
|
|
return result
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to perform batch update: {str(e)}")
|
|
raise
|
|
|
|
async def format_header_row(
|
|
self,
|
|
spreadsheet_id: str,
|
|
sheet_name: str,
|
|
sheet_id: int
|
|
) -> dict:
|
|
"""
|
|
Format the first row as a header (bold, frozen).
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab
|
|
sheet_id: Internal sheet ID (different from sheet_name)
|
|
|
|
Returns:
|
|
Dictionary with formatting result
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
requests = [
|
|
# Make first row bold
|
|
{
|
|
'repeatCell': {
|
|
'range': {
|
|
'sheetId': sheet_id,
|
|
'startRowIndex': 0,
|
|
'endRowIndex': 1
|
|
},
|
|
'cell': {
|
|
'userEnteredFormat': {
|
|
'textFormat': {
|
|
'bold': True
|
|
},
|
|
'backgroundColor': {
|
|
'red': 0.9,
|
|
'green': 0.9,
|
|
'blue': 0.9
|
|
}
|
|
}
|
|
},
|
|
'fields': 'userEnteredFormat(textFormat,backgroundColor)'
|
|
}
|
|
},
|
|
# Freeze first row
|
|
{
|
|
'updateSheetProperties': {
|
|
'properties': {
|
|
'sheetId': sheet_id,
|
|
'gridProperties': {
|
|
'frozenRowCount': 1
|
|
}
|
|
},
|
|
'fields': 'gridProperties.frozenRowCount'
|
|
}
|
|
}
|
|
]
|
|
|
|
result = await self.batch_update(spreadsheet_id, requests)
|
|
logger.info(f"Successfully formatted header row for sheet '{sheet_name}'")
|
|
return result
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to format header row: {str(e)}")
|
|
raise
|
|
|
|
def get_sheet_id(self, spreadsheet_id: str, sheet_name: str) -> Optional[int]:
|
|
"""
|
|
Get the internal sheet ID for a given sheet name.
|
|
|
|
Args:
|
|
spreadsheet_id: ID of the Google Sheets document
|
|
sheet_name: Name of the sheet tab
|
|
|
|
Returns:
|
|
Internal sheet ID or None if not found
|
|
"""
|
|
if not self.service:
|
|
raise RuntimeError("Google Sheets client not authenticated")
|
|
|
|
try:
|
|
spreadsheet = self.service.spreadsheets().get(
|
|
spreadsheetId=spreadsheet_id
|
|
).execute()
|
|
|
|
sheets = spreadsheet.get('sheets', [])
|
|
for sheet in sheets:
|
|
if sheet['properties']['title'] == sheet_name:
|
|
return sheet['properties']['sheetId']
|
|
|
|
return None
|
|
|
|
except HttpError as e:
|
|
logger.error(f"Failed to get sheet ID: {str(e)}")
|
|
return None
|
|
|