chore(sync): Task 3 review fixes——删未用 import + module docstring 诚实化 + repr(exc) 一致 + 批处理 guard
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -32,7 +32,10 @@ sync continues (the message is still upserted; the watermark still advances).
|
|||||||
Transactions: the orchestrator never commits. Every write is flushed into the
|
Transactions: the orchestrator never commits. Every write is flushed into the
|
||||||
caller's ``AsyncSession``; the caller owns the commit / rollback. This keeps a
|
caller's ``AsyncSession``; the caller owns the commit / rollback. This keeps a
|
||||||
multi-entity batch atomic at the caller's discretion and lets a test roll a
|
multi-entity batch atomic at the caller's discretion and lets a test roll a
|
||||||
failed advance back by simply not committing.
|
failed advance back by simply not committing. Exception:
|
||||||
|
:func:`sync_call_recordings` commits its own ``SyncRun`` because
|
||||||
|
:func:`download_call_recordings` commits per batch internally, so the run
|
||||||
|
bookkeeping must land in the same transaction boundary.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
@@ -609,7 +612,11 @@ async def sync_call_recordings(
|
|||||||
caller), this one commits its own ``SyncRun`` row: the underlying
|
caller), this one commits its own ``SyncRun`` row: the underlying
|
||||||
:func:`download_call_recordings` commits per batch as it advances, so the
|
:func:`download_call_recordings` commits per batch as it advances, so the
|
||||||
run bookkeeping must land in the same transaction boundary to stay
|
run bookkeeping must land in the same transaction boundary to stay
|
||||||
consistent with what was actually persisted.
|
consistent with what was actually persisted. Because it commits internally,
|
||||||
|
it must NOT be added to a batched runner (:func:`_run_all` /
|
||||||
|
:func:`sync_all_wechat`) without reconsidering per-batch atomicity — a
|
||||||
|
batched wrapper can no longer treat the whole entity set as one rollback
|
||||||
|
unit once this func has already persisted its own rows.
|
||||||
|
|
||||||
Failure policy mirrors every other ``sync_*``: a downloader exception is
|
Failure policy mirrors every other ``sync_*``: a downloader exception is
|
||||||
recorded as a ``status='failed'`` ``SyncRun`` and a failed
|
recorded as a ``status='failed'`` ``SyncRun`` and a failed
|
||||||
@@ -634,7 +641,7 @@ async def sync_call_recordings(
|
|||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.exception("sync_call_recordings failed")
|
logger.exception("sync_call_recordings failed")
|
||||||
await _finish_run(
|
await _finish_run(
|
||||||
session, run, status="failed", fetched=0, upserted=0, error=str(exc)
|
session, run, status="failed", fetched=0, upserted=0, error=repr(exc)
|
||||||
)
|
)
|
||||||
await session.commit()
|
await session.commit()
|
||||||
return SyncResult(
|
return SyncResult(
|
||||||
@@ -642,7 +649,7 @@ async def sync_call_recordings(
|
|||||||
status="failed",
|
status="failed",
|
||||||
fetched=0,
|
fetched=0,
|
||||||
upserted=0,
|
upserted=0,
|
||||||
error=str(exc),
|
error=repr(exc),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -29,7 +29,6 @@ from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
|||||||
|
|
||||||
from app.db.base import Base
|
from app.db.base import Base
|
||||||
from app.db.models import PhoneCall, SyncRun
|
from app.db.models import PhoneCall, SyncRun
|
||||||
from app.scraper.media import download_call_recordings
|
|
||||||
from app.session.yunke_client import YunkeClient
|
from app.session.yunke_client import YunkeClient
|
||||||
from app.storage.object_store import ObjectStore
|
from app.storage.object_store import ObjectStore
|
||||||
from app.sync import orchestrator
|
from app.sync import orchestrator
|
||||||
|
|||||||
Reference in New Issue
Block a user