"""Google Sheets service — business logic for the g_sheet domain. The Google client is blocking, so every call goes through asyncio.to_thread rather than stalling the event loop. Client construction is lazy and guarded by a lock so concurrent requests build it exactly once. """ import asyncio import logging import threading from datetime import datetime,timezone from fastapi import HTTPException from g_sheet.plugins import ( SCOPES, SPREADSHEET_ID, SPREADSHEET_NAME, SPREADSHEET_URL, SheetsServiceError, build_sheets_client, ensure_fresh, execute, import_row_stats, load_credentials, map_record_to_form_data, normalise_headers, quote_tab, rows_to_indexed_records, rows_to_records, stringify_rows, ) from g_sheet.models import FormData,SheetImportRun from g_sheet.serializers import ( serialize_append, serialize_clear, serialize_form_data, serialize_health, serialize_import, serialize_import_all, serialize_import_run, serialize_metadata, serialize_records, serialize_sheet_summary, serialize_update, serialize_values, ) logger=logging.getLogger("g_sheet.views") class Sheet: def __init__(self,session=None,spreadsheet_id=None,credentials_path=None,scopes=None): self.session=session self.spreadsheet_id=spreadsheet_id or SPREADSHEET_ID self.spreadsheet_name=SPREADSHEET_NAME self.spreadsheet_url=SPREADSHEET_URL self.credentials_path=credentials_path self.scopes=scopes or SCOPES self.credentials=None self.client=None self._lock=threading.Lock() def _require_session(self): if self.session is None: raise HTTPException(status_code=500,detail="Database session is required") return self.session # -- client ------------------------------------------------------------ def _connect(self): """Build credentials + client once, then keep refreshing the same token. Double-checked under the lock: two requests racing here must not each build their own client. """ if self.client is not None: return ensure_fresh(self.credentials) and self.client with self._lock: if self.client is None: self.credentials=load_credentials(self.credentials_path,self.scopes) self.client=build_sheets_client(self.credentials) else: ensure_fresh(self.credentials) return self.client async def _values(self): if not self.spreadsheet_id: raise HTTPException(status_code=500,detail="SPREADSHEET_ID is not configured") client=await asyncio.to_thread(self._connect) return client.spreadsheets().values() async def _spreadsheets(self): if not self.spreadsheet_id: raise HTTPException(status_code=500,detail="SPREADSHEET_ID is not configured") client=await asyncio.to_thread(self._connect) return client.spreadsheets() # -- reads ------------------------------------------------------------- async def get_metadata(self): """Spreadsheet title, id, url and every tab with its row/column counts.""" try: spreadsheets=await self._spreadsheets() request=spreadsheets.get(spreadsheetId=self.spreadsheet_id,fields=( "spreadsheetId,spreadsheetUrl,properties(title,locale,timeZone)," "sheets(properties(sheetId,title,index,gridProperties(rowCount,columnCount)))" )) payload=await asyncio.to_thread(execute,request,"spreadsheet metadata") return serialize_metadata(payload) except SheetsServiceError as e: raise HTTPException(status_code=e.status_code,detail=e.message) async def list_tabs(self): """Tab titles in sheet order.""" metadata=await self.get_metadata() return [tab["title"] for tab in metadata["tabs"] if tab.get("title")] async def read_range(self,tab,cell_range=None): """Raw rows for a tab, or for a sub-range of it when cell_range is given.""" try: values=await self._values() target=quote_tab(tab,cell_range) request=values.get(spreadsheetId=self.spreadsheet_id,range=target) payload=await asyncio.to_thread(execute,request,f"read {target}") rows=stringify_rows(payload.get("values")) return serialize_values(tab,cell_range,rows) except SheetsServiceError as e: raise HTTPException(status_code=e.status_code,detail=e.message) async def read_records(self,tab): """Rows keyed by the first row. Blank rows are skipped, short rows padded.""" data=await self.read_range(tab) return serialize_records(tab,rows_to_records(data["rows"])) async def read_all(self): """Every tab as records, keyed by tab name.""" tabs=await self.list_tabs() sheets={} for tab in tabs: data=await self.read_records(tab) sheets[tab]=data["records"] return {"sheets":sheets,"tabs":tabs,"total":len(tabs)} # -- writes ------------------------------------------------------------ async def append_rows(self,tab,rows): """Append rows below the tab's current content.""" if not rows: raise HTTPException(status_code=422,detail="rows must not be empty") try: values=await self._values() target=quote_tab(tab) request=values.append( spreadsheetId=self.spreadsheet_id, range=target, valueInputOption="USER_ENTERED", insertDataOption="INSERT_ROWS", body={"values":rows}, ) payload=await asyncio.to_thread(execute,request,f"append to {target}") return serialize_append(tab,payload) except SheetsServiceError as e: raise HTTPException(status_code=e.status_code,detail=e.message) async def update_range(self,tab,cell_range,rows): """Overwrite an explicit A1 range with rows.""" if not cell_range: raise HTTPException(status_code=422,detail="cell_range is required") if not rows: raise HTTPException(status_code=422,detail="rows must not be empty") try: values=await self._values() target=quote_tab(tab,cell_range) request=values.update( spreadsheetId=self.spreadsheet_id, range=target, valueInputOption="USER_ENTERED", body={"values":rows}, ) payload=await asyncio.to_thread(execute,request,f"update {target}") return serialize_update(tab,payload) except SheetsServiceError as e: raise HTTPException(status_code=e.status_code,detail=e.message) async def clear_range(self,tab,cell_range): """Clear the values in an explicit A1 range, leaving formatting intact.""" if not cell_range: raise HTTPException(status_code=422,detail="cell_range is required") try: values=await self._values() target=quote_tab(tab,cell_range) request=values.clear(spreadsheetId=self.spreadsheet_id,range=target,body={}) payload=await asyncio.to_thread(execute,request,f"clear {target}") return serialize_clear(tab,payload) except SheetsServiceError as e: raise HTTPException(status_code=e.status_code,detail=e.message) # -- FormData import / query ------------------------------------------- async def import_sheet(self,tab): """Read one tab from Google Sheets and replace its FormData rows.""" session=self._require_session() if not tab or not str(tab).strip(): raise HTTPException(status_code=422,detail="tab is required") tab=str(tab).strip() data=await self.read_range(tab) rows=data["rows"] if not rows: return serialize_import({"tab":tab,"rows_read":0,"inserted":0,"deleted":0}) headers=normalise_headers(rows[0]) indexed=rows_to_indexed_records(rows) mapped=[ map_record_to_form_data(tab,record,headers,row_number) for row_number,record in indexed ] result=await FormData.replace_sheet(session,tab,mapped) stats=import_row_stats(mapped,headers) return serialize_import({ "tab":tab, "rows_read":len(indexed), "inserted":result["inserted"], "deleted":result["deleted"], **stats, }) async def import_all(self): """Import every tab sequentially; one tab failure does not abort the rest.""" self._require_session() tabs=await self.list_tabs() reports=[] for tab in tabs: try: report=await self.import_sheet(tab) reports.append(report) except HTTPException as e: logger.warning("import_all tab %s failed: %s",tab,e.detail) reports.append(serialize_import({ "tab":tab,"rows_read":0,"inserted":0,"deleted":0, "error":str(e.detail), })) except Exception as e: logger.exception("import_all tab %s failed",tab) reports.append(serialize_import({ "tab":tab,"rows_read":0,"inserted":0,"deleted":0, "error":str(e), })) return serialize_import_all(reports) async def get_form_data(self,sheet=None,search=None,top=None,skip=None): session=self._require_session() rows=await FormData.fetch_form_data( session,sheet=sheet,search=search,top=top,skip=skip, ) total=await FormData.count_form_data(session,sheet=sheet,search=search) return [serialize_form_data(row) for row in rows],total async def get_form_data_by_id(self,record_id): session=self._require_session() row=await FormData.get_form_data_by_id(session,record_id) if not row: raise HTTPException(status_code=404,detail="Form data not found") return serialize_form_data(row) async def get_imported_sheets(self): session=self._require_session() sheets=await FormData.get_sheet_names(session) return serialize_sheet_summary(sheets) async def delete_sheet_data(self,tab): session=self._require_session() if not tab or not str(tab).strip(): raise HTTPException(status_code=422,detail="tab is required") deleted=await FormData.delete_by_sheet(session,str(tab).strip()) return {"tab":str(tab).strip(),"deleted":deleted} async def start_import(self,current_user=None,tab=None): """Enqueue a sheet import on the shared Taskiq worker; return the run row. If a queued/running import already exists, return it instead of stacking another. """ session=self._require_session() active=await SheetImportRun.get_active(session) if active: return serialize_import_run(active) created_by=None if isinstance(current_user,dict) and current_user.get("id"): created_by=SheetImportRun._as_uuid(current_user.get("id")) tab_value=str(tab).strip() if tab else None row=await SheetImportRun.insert_run(session,{ "status":"queued", "created_by":created_by, "tab":tab_value, }) from g_sheet.tasks import import_sheets from taskiq_management.g_sheet_broker_setup import SHEET_QUEUE_NAME task=await import_sheets.kicker().with_labels( created_at=datetime.now(timezone.utc).isoformat(), correlation_id=str(row.id), queue=SHEET_QUEUE_NAME, ).kiq(str(row.id)) row=await SheetImportRun.update_run(session,row.id,{"task_id":task.task_id}) return serialize_import_run(row) async def get_import_run(self,run_id=None): session=self._require_session() if run_id: row=await SheetImportRun.get_by_id(session,run_id) if not row: raise HTTPException(status_code=404,detail="Import run not found") return serialize_import_run(row) row=await SheetImportRun.get_active(session) if row: return serialize_import_run(row) from sqlmodel import select result=await session.execute( select(SheetImportRun).order_by(SheetImportRun.created_at.desc()).limit(1) ) row=result.scalars().first() if not row: raise HTTPException(status_code=404,detail="No import runs yet") return serialize_import_run(row) # -- health ------------------------------------------------------------ async def health_check(self): """Credentials + sheet reachability as a status dict. Never raises.""" if not self.spreadsheet_id: return serialize_health(False,"SPREADSHEET_ID is not configured") try: tabs=await self.list_tabs() return serialize_health(True,"spreadsheet reachable",tabs) except HTTPException as e: logger.warning("sheets health check failed: %s",e.detail) return serialize_health(False,str(e.detail)) except Exception as e: logger.warning("sheets health check failed: %s",e) return serialize_health(False,str(e))