From 32c737fa989f8c9f222afa6c91e112d3009650c1 Mon Sep 17 00:00:00 2001 From: Donkie Date: Fri, 16 Aug 2024 15:13:09 +0200 Subject: [PATCH] Revert so commit now happens before notice, but add a catch-all to websocket notifications --- spoolman/database/filament.py | 28 ++++++++++++++----------- spoolman/database/spool.py | 39 +++++++++++++++++++---------------- spoolman/database/vendor.py | 29 ++++++++++++++++---------- 3 files changed, 55 insertions(+), 41 deletions(-) diff --git a/spoolman/database/filament.py b/spoolman/database/filament.py index 73ba448..2613e72 100644 --- a/spoolman/database/filament.py +++ b/spoolman/database/filament.py @@ -76,8 +76,8 @@ async def create( extra=[models.FilamentField(key=k, value=v) for k, v in (extra or {}).items()], ) db.add(filament) - await filament_changed(filament, EventType.ADDED) await db.commit() + await filament_changed(filament, EventType.ADDED) return filament @@ -172,8 +172,8 @@ async def update( filament.multi_color_direction = v.value if v is not None else None else: setattr(filament, k, v) - await filament_changed(filament, EventType.UPDATED) await db.commit() + await filament_changed(filament, EventType.UPDATED) return filament @@ -182,8 +182,8 @@ async def delete(db: AsyncSession, filament_id: int) -> None: filament = await get_by_id(db, filament_id) await db.delete(filament) try: - await filament_changed(filament, EventType.DELETED) await db.commit() # Flush immediately so any errors are propagated in this request. + await filament_changed(filament, EventType.DELETED) except IntegrityError as exc: await db.rollback() raise ItemDeleteError("Failed to delete filament.") from exc @@ -255,12 +255,16 @@ async def find_by_color( async def filament_changed(filament: models.Filament, typ: EventType) -> None: """Notify websocket clients that a filament has changed.""" - await websocket_manager.send( - ("filament", str(filament.id)), - FilamentEvent( - type=typ, - resource="filament", - date=datetime.utcnow(), - payload=Filament.from_db(filament), - ), - ) + try: + await websocket_manager.send( + ("filament", str(filament.id)), + FilamentEvent( + type=typ, + resource="filament", + date=datetime.utcnow(), + payload=Filament.from_db(filament), + ), + ) + except Exception: + # Important to have a catch-all here since we don't want to stop the call if this fails. + logger.exception("Failed to send websocket message") diff --git a/spoolman/database/spool.py b/spoolman/database/spool.py index 594d392..b70aac3 100644 --- a/spoolman/database/spool.py +++ b/spoolman/database/spool.py @@ -1,5 +1,6 @@ """Helper functions for interacting with spool database objects.""" +import logging from collections.abc import Sequence from datetime import datetime, timezone from typing import Optional, Union @@ -25,6 +26,8 @@ from spoolman.exceptions import ItemCreateError, ItemNotFoundError, SpoolMeasure from spoolman.math import weight_from_length from spoolman.ws import websocket_manager +logger = logging.getLogger(__name__) + def utc_timezone_naive(dt: datetime) -> datetime: """Convert a datetime object to UTC and remove timezone info.""" @@ -92,8 +95,8 @@ async def create( extra=[models.SpoolField(key=k, value=v) for k, v in (extra or {}).items()], ) db.add(spool) - await spool_changed(spool, EventType.ADDED) await db.commit() + await spool_changed(spool, EventType.ADDED) return spool @@ -228,8 +231,8 @@ async def update( spool.extra = [models.SpoolField(key=k, value=v) for k, v in v.items()] else: setattr(spool, k, v) - await spool_changed(spool, EventType.UPDATED) await db.commit() + await spool_changed(spool, EventType.UPDATED) return spool @@ -291,8 +294,8 @@ async def use_weight(db: AsyncSession, spool_id: int, weight: float) -> models.S spool.first_used = datetime.utcnow().replace(microsecond=0) spool.last_used = datetime.utcnow().replace(microsecond=0) - await spool_changed(spool, EventType.UPDATED) await db.commit() + await spool_changed(spool, EventType.UPDATED) return spool @@ -337,12 +340,8 @@ async def use_length(db: AsyncSession, spool_id: int, length: float) -> models.S spool.first_used = datetime.utcnow().replace(microsecond=0) spool.last_used = datetime.utcnow().replace(microsecond=0) - await spool_changed(spool, EventType.UPDATED) - - # Commit should be the last action, everything after that must never fail - # Otherwise you can end up in a non-atomic thing where the http use request fails - # but the data still has been committed. await db.commit() + await spool_changed(spool, EventType.UPDATED) return spool @@ -436,15 +435,19 @@ async def find_lot_numbers( async def spool_changed(spool: models.Spool, typ: EventType) -> None: """Notify websocket clients that a spool has changed.""" - await websocket_manager.send( - ("spool", str(spool.id)), - SpoolEvent( - type=typ, - resource="spool", - date=datetime.utcnow(), - payload=Spool.from_db(spool), - ), - ) + try: + await websocket_manager.send( + ("spool", str(spool.id)), + SpoolEvent( + type=typ, + resource="spool", + date=datetime.utcnow(), + payload=Spool.from_db(spool), + ), + ) + except Exception: + # Important to have a catch-all here since we don't want to stop the call if this fails. + logger.exception("Failed to send websocket message") async def reset_initial_weight(db: AsyncSession, spool_id: int, weight: float) -> models.Spool: @@ -453,6 +456,6 @@ async def reset_initial_weight(db: AsyncSession, spool_id: int, weight: float) - spool.initial_weight = weight spool.used_weight = 0 - await spool_changed(spool, EventType.UPDATED) await db.commit() + await spool_changed(spool, EventType.UPDATED) return spool diff --git a/spoolman/database/vendor.py b/spoolman/database/vendor.py index 3d07ba0..a348e48 100644 --- a/spoolman/database/vendor.py +++ b/spoolman/database/vendor.py @@ -1,5 +1,6 @@ """Helper functions for interacting with vendor database objects.""" +import logging from datetime import datetime from typing import Optional @@ -13,6 +14,8 @@ from spoolman.database.utils import SortOrder, add_where_clause_str, add_where_c from spoolman.exceptions import ItemNotFoundError from spoolman.ws import websocket_manager +logger = logging.getLogger(__name__) + async def create( *, @@ -33,8 +36,8 @@ async def create( extra=[models.VendorField(key=k, value=v) for k, v in (extra or {}).items()], ) db.add(vendor) - await vendor_changed(vendor, EventType.ADDED) await db.commit() + await vendor_changed(vendor, EventType.ADDED) return vendor @@ -101,8 +104,8 @@ async def update( vendor.extra = [models.VendorField(key=k, value=v) for k, v in v.items()] else: setattr(vendor, k, v) - await vendor_changed(vendor, EventType.UPDATED) await db.commit() + await vendor_changed(vendor, EventType.UPDATED) return vendor @@ -122,12 +125,16 @@ async def clear_extra_field(db: AsyncSession, key: str) -> None: async def vendor_changed(vendor: models.Vendor, typ: EventType) -> None: """Notify websocket clients that a vendor has changed.""" - await websocket_manager.send( - ("vendor", str(vendor.id)), - VendorEvent( - type=typ, - resource="vendor", - date=datetime.utcnow(), - payload=Vendor.from_db(vendor), - ), - ) + try: + await websocket_manager.send( + ("vendor", str(vendor.id)), + VendorEvent( + type=typ, + resource="vendor", + date=datetime.utcnow(), + payload=Vendor.from_db(vendor), + ), + ) + except Exception: + # Important to have a catch-all here since we don't want to stop the call if this fails. + logger.exception("Failed to send websocket message")