fix(backend): découpe l'insert des alertes en lots de 1000
`AlertRepository.create_many` envoyait un `INSERT` d'un seul tenant. À douze colonnes par alerte, le plafond asyncpg de 32 767 paramètres tombe à 2 730 lignes : une détection sur une fenêtre chargée échouait en `InterfaceError`, et le DAG `alertes` avec elle. Reprend le patron déjà en place dans `RecommendationRepository.create_missing`.
This commit is contained in:
@@ -6,6 +6,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||||||
|
|
||||||
from app.models.energy import Alert
|
from app.models.energy import Alert
|
||||||
|
|
||||||
|
# Douze colonnes par alerte, contre quatre pour une recommandation : le plafond asyncpg de
|
||||||
|
# 32 767 parametres tombe a 2 730 lignes, d'ou un lot plus petit que `recommendation.py`.
|
||||||
|
TAILLE_DE_LOT = 1000
|
||||||
|
|
||||||
|
|
||||||
class AlertRepository:
|
class AlertRepository:
|
||||||
def __init__(self, session: AsyncSession) -> None:
|
def __init__(self, session: AsyncSession) -> None:
|
||||||
@@ -44,12 +48,17 @@ class AlertRepository:
|
|||||||
}
|
}
|
||||||
for alerte in alerts
|
for alerte in alerts
|
||||||
]
|
]
|
||||||
requete = (
|
creees: list[Alert] = []
|
||||||
insert(Alert)
|
# Piège : asyncpg plafonne une requête à 32 767 paramètres. Une détection sur une fenêtre
|
||||||
.values(valeurs)
|
# chargée dépasse ce seuil, et l'`INSERT` d'un seul tenant échouerait.
|
||||||
.on_conflict_do_nothing(constraint="uq_alert_source_reference")
|
for debut in range(0, len(valeurs), TAILLE_DE_LOT):
|
||||||
.returning(Alert)
|
requete = (
|
||||||
)
|
insert(Alert)
|
||||||
resultat = await self._session.execute(requete)
|
.values(valeurs[debut : debut + TAILLE_DE_LOT])
|
||||||
|
.on_conflict_do_nothing(constraint="uq_alert_source_reference")
|
||||||
|
.returning(Alert)
|
||||||
|
)
|
||||||
|
resultat = await self._session.execute(requete)
|
||||||
|
creees.extend(resultat.scalars().all())
|
||||||
await self._session.flush()
|
await self._session.flush()
|
||||||
return resultat.scalars().all()
|
return creees
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import pytest
|
|||||||
from sqlalchemy.ext.asyncio import AsyncSession
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
from app.models.energy import Alert
|
from app.models.energy import Alert
|
||||||
|
from app.repositories import alert as module_alert
|
||||||
from app.repositories.alert import AlertRepository
|
from app.repositories.alert import AlertRepository
|
||||||
from app.schemas.alert import AlertSeverity
|
from app.schemas.alert import AlertSeverity
|
||||||
from tests.repositories.test_site import creer as creer_site
|
from tests.repositories.test_site import creer as creer_site
|
||||||
@@ -146,3 +147,20 @@ async def test_create_many_does_nothing_for_an_empty_list(session: AsyncSession)
|
|||||||
creees = await depot.create_many([])
|
creees = await depot.create_many([])
|
||||||
|
|
||||||
assert creees == []
|
assert creees == []
|
||||||
|
|
||||||
|
|
||||||
|
async def test_create_many_inserts_every_alert_across_several_batches(
|
||||||
|
session: AsyncSession, monkeypatch: pytest.MonkeyPatch
|
||||||
|
) -> None:
|
||||||
|
monkeypatch.setattr(module_alert, "TAILLE_DE_LOT", 2)
|
||||||
|
site = await creer_site(session)
|
||||||
|
depot = AlertRepository(session)
|
||||||
|
a_inserer = [
|
||||||
|
_alerte_a_inserer(site_id=site.site_id, source_alert_id=f"threshold:lot-{index}")
|
||||||
|
for index in range(5)
|
||||||
|
]
|
||||||
|
|
||||||
|
creees = await depot.create_many(a_inserer)
|
||||||
|
await session.rollback()
|
||||||
|
|
||||||
|
assert len(creees) == 5
|
||||||
|
|||||||
Reference in New Issue
Block a user