Refactor Google Sheets service integration
- Updated the `app.py` file to replace the generic `Sheet` service with specific services: `SheetHealth`, `SheetRead`, `SheetImport`, and `SheetWrite` for better clarity and functionality. - Modified the `tasks.py` file to utilize the `SheetImport` service for handling sheet import tasks. - Enhanced the `views.py` file by introducing a new class hierarchy for sheet operations, improving code organization and readability. These changes improve the structure and maintainability of the Google Sheets integration.pull/29/head
parent
6f52436165
commit
d5879a0617
|
|
@ -4,7 +4,13 @@ from pydantic import BaseModel
|
|||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from db_setup import get_session
|
||||
from g_sheet.views import Sheet
|
||||
from g_sheet.views import (
|
||||
SheetFormData,
|
||||
SheetHealth,
|
||||
SheetImport,
|
||||
SheetRead,
|
||||
SheetWrite,
|
||||
)
|
||||
from users.permissions import PermissionTag,require_permission
|
||||
from dotenv import load_dotenv
|
||||
load_dotenv()
|
||||
|
|
@ -33,7 +39,7 @@ async def sheet_health():
|
|||
comes back as {"status":"error"} so a probe can read the reason.
|
||||
"""
|
||||
try:
|
||||
service=Sheet()
|
||||
service=SheetHealth()
|
||||
data=await service.health_check()
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -48,7 +54,7 @@ async def fetch_sheet_metadata(
|
|||
current_user: dict = Depends(require_permission(PermissionTag.SETTINGS_VIEW)),
|
||||
):
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetRead(spreadsheet_id=spreadsheet_id)
|
||||
data=await service.get_metadata()
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -63,7 +69,7 @@ async def fetch_sheet_tabs(
|
|||
current_user: dict = Depends(require_permission(PermissionTag.SETTINGS_VIEW)),
|
||||
):
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetRead(spreadsheet_id=spreadsheet_id)
|
||||
items=await service.list_tabs()
|
||||
return JSONResponse(content={"data":items,"total":len(items),"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -83,7 +89,7 @@ async def fetch_sheet(
|
|||
"""No tab -> every tab as records. With a tab -> that tab, header-mapped unless
|
||||
raw=true, which returns the rows exactly as the sheet stores them."""
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetRead(spreadsheet_id=spreadsheet_id)
|
||||
if not tab:
|
||||
data=await service.read_all()
|
||||
return JSONResponse(content={"data":data["sheets"],"total":data["total"],"status_code":200})
|
||||
|
|
@ -106,7 +112,7 @@ async def import_all_sheets(
|
|||
):
|
||||
"""No tab -> every tab. With a tab -> that sheet only. Poll GET /sheet/import/fetch."""
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetImport(session=session)
|
||||
data=await service.start_import(current_user=current_user,tab=tab)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -123,7 +129,7 @@ async def import_one_sheet(
|
|||
):
|
||||
"""Enqueue a single-tab import. Poll GET /sheet/import/fetch for status."""
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetImport(session=session)
|
||||
data=await service.start_import(current_user=current_user,tab=tab)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -139,7 +145,7 @@ async def fetch_sheet_import(
|
|||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetImport(session=session)
|
||||
data=await service.get_import_run(run_id=run_id)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -154,7 +160,7 @@ async def fetch_form_data_sheets(
|
|||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetFormData(session=session)
|
||||
data=await service.get_imported_sheets()
|
||||
return JSONResponse(content={"data":data,"total":data["total"],"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -173,7 +179,7 @@ async def fetch_form_data(
|
|||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetFormData(session=session)
|
||||
items,total=await service.get_form_data(sheet=sheet,search=search,top=top,skip=skip)
|
||||
return JSONResponse(content={"data":items,"total":total,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -189,7 +195,7 @@ async def fetch_form_data_by_id(
|
|||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetFormData(session=session)
|
||||
data=await service.get_form_data_by_id(record_id)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -205,7 +211,7 @@ async def delete_form_data_sheet(
|
|||
session: AsyncSession = Depends(get_session),
|
||||
):
|
||||
try:
|
||||
service=Sheet(session=session)
|
||||
service=SheetFormData(session=session)
|
||||
data=await service.delete_sheet_data(tab)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -222,7 +228,7 @@ async def append_sheet_rows(
|
|||
spreadsheet_id: str | None = Query(None),
|
||||
):
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetWrite(spreadsheet_id=spreadsheet_id)
|
||||
data=await service.append_rows(tab,payload.rows)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -239,7 +245,7 @@ async def update_sheet_range(
|
|||
spreadsheet_id: str | None = Query(None),
|
||||
):
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetWrite(spreadsheet_id=spreadsheet_id)
|
||||
data=await service.update_range(tab,payload.cell_range,payload.rows)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
@ -256,7 +262,7 @@ async def clear_sheet_range(
|
|||
spreadsheet_id: str | None = Query(None),
|
||||
):
|
||||
try:
|
||||
service=Sheet(spreadsheet_id=spreadsheet_id)
|
||||
service=SheetWrite(spreadsheet_id=spreadsheet_id)
|
||||
data=await service.clear_range(tab,payload.cell_range)
|
||||
return JSONResponse(content={"data":data,"total":1,"status_code":200})
|
||||
except HTTPException:
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ from dotenv import load_dotenv
|
|||
|
||||
from db_setup import session_scope
|
||||
from g_sheet.models import SheetImportRun
|
||||
from g_sheet.views import Sheet
|
||||
from g_sheet.views import SheetImport
|
||||
from taskiq_management.broker_setup import MAX_RETRIES,RETRY_DELAY
|
||||
from taskiq_management.g_sheet_broker_setup import sheet_broker
|
||||
from taskiq_management.middleware import PermanentTaskError
|
||||
|
|
@ -64,7 +64,7 @@ async def import_sheets(run_id:str) -> dict:
|
|||
tab=row.tab
|
||||
|
||||
async with session_scope() as session:
|
||||
service=Sheet(session=session)
|
||||
service=SheetImport(session=session)
|
||||
try:
|
||||
if tab:
|
||||
report=await service.import_sheet(tab)
|
||||
|
|
|
|||
|
|
@ -3,6 +3,15 @@
|
|||
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.
|
||||
|
||||
Hierarchy:
|
||||
Sheet shared config / session
|
||||
└─ SheetClient credentials + spreadsheets client
|
||||
├─ SheetRead
|
||||
│ ├─ SheetHealth
|
||||
│ └─ SheetImport
|
||||
└─ SheetWrite
|
||||
SheetFormData DB mirror only (no Google client)
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
|
@ -50,6 +59,8 @@ logger=logging.getLogger("g_sheet.views")
|
|||
|
||||
|
||||
class Sheet:
|
||||
"""Parent: spreadsheet identity, optional DB session, and shared helpers."""
|
||||
|
||||
def __init__(self,session=None,spreadsheet_id=None,credentials_path=None,scopes=None):
|
||||
self.session=session
|
||||
self.spreadsheet_id=spreadsheet_id or SPREADSHEET_ID
|
||||
|
|
@ -66,7 +77,9 @@ class Sheet:
|
|||
raise HTTPException(status_code=500,detail="Database session is required")
|
||||
return self.session
|
||||
|
||||
# -- client ------------------------------------------------------------
|
||||
|
||||
class SheetClient(Sheet):
|
||||
"""Google API client — lazy connect, token refresh, values/spreadsheets handles."""
|
||||
|
||||
def _connect(self):
|
||||
"""Build credentials + client once, then keep refreshing the same token.
|
||||
|
|
@ -96,7 +109,9 @@ class Sheet:
|
|||
client=await asyncio.to_thread(self._connect)
|
||||
return client.spreadsheets()
|
||||
|
||||
# -- reads -------------------------------------------------------------
|
||||
|
||||
class SheetRead(SheetClient):
|
||||
"""Read-only sheet operations."""
|
||||
|
||||
async def get_metadata(self):
|
||||
"""Spreadsheet title, id, url and every tab with its row/column counts."""
|
||||
|
|
@ -142,7 +157,9 @@ class Sheet:
|
|||
sheets[tab]=data["records"]
|
||||
return {"sheets":sheets,"tabs":tabs,"total":len(tabs)}
|
||||
|
||||
# -- writes ------------------------------------------------------------
|
||||
|
||||
class SheetWrite(SheetClient):
|
||||
"""Mutating sheet operations."""
|
||||
|
||||
async def append_rows(self,tab,rows):
|
||||
"""Append rows below the tab's current content."""
|
||||
|
|
@ -196,7 +213,27 @@ class Sheet:
|
|||
except SheetsServiceError as e:
|
||||
raise HTTPException(status_code=e.status_code,detail=e.message)
|
||||
|
||||
# -- FormData import / query -------------------------------------------
|
||||
|
||||
class SheetHealth(SheetRead):
|
||||
"""Credentials + spreadsheet reachability."""
|
||||
|
||||
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))
|
||||
|
||||
|
||||
class SheetImport(SheetRead):
|
||||
"""Google Sheet → FormData import + import-run tracking."""
|
||||
|
||||
async def import_sheet(self,tab):
|
||||
"""Read one tab from Google Sheets and replace its FormData rows."""
|
||||
|
|
@ -247,33 +284,6 @@ class Sheet:
|
|||
}))
|
||||
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.
|
||||
|
||||
|
|
@ -324,18 +334,33 @@ class Sheet:
|
|||
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))
|
||||
class SheetFormData(Sheet):
|
||||
"""FormData DB mirror — query / delete only (no Google client)."""
|
||||
|
||||
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}
|
||||
|
|
|
|||
Loading…
Reference in New Issue