Migrate to SQLAlchemy v2 API

le717 · Jun 8, 2024 · 193e792 · 193e792
1 parent 5d3dcc6
commit 193e792
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 19 deletions.
diff --git a/.vscode/launch.json b/.vscode/launch.json
@@ -18,7 +18,7 @@
         "FLASK_DEBUG": "1",
         "SYS_VARS_PATH": "${workspaceFolder}/secrets",
         "DB_PATH": "${workspaceFolder}/database/database.db",
-        "TIMES_FAILED_THRESHOLD": "2",
+        "TIMES_FAILED_THRESHOLD": "3",
         "ENABLE_DISCORD_LOGGING": "false"
       },
       "args": [

diff --git a/CHANGES.md b/CHANGES.md
@@ -16,8 +16,7 @@ _Released TDB_
 - Update OpenAPI spec version
 - Update minimum Python version to 3.11
 - Update to Flask v3
-- Update to SQLAlchemy v2
-- ~~Migrate SQLAlchemy to v2 API~~ _update pending_
+- Update to SQLAlchemy v2 and API
 - Update Docker image to `python:3.11-slim`
 - Update all dependencies to their latest versions
 - Switch to [ruff](https://docs.astral.sh/ruff/) for linting and formatting

diff --git a/src/core/database/linkrot.py b/src/core/database/linkrot.py
@@ -32,8 +32,12 @@ def __ping_url(url: str) -> bool:
             httpx.codes.NO_CONTENT,
             httpx.codes.NOT_MODIFIED,
         )
-    except httpx.HTTPError as exc:
-        logger.exception("An error occurred when checking a URL for rotting.", exc_info=exc)
+    except httpx.HTTPError:
+        logger.info({
+            "id": "N/A",
+            "url": url,
+            "message": "Link could not be reached during a linkrot check.",
+        })
         return False
 
 
@@ -54,13 +58,15 @@ def __create(data: WebLink) -> Literal[True]:
 
 
 def __get(uuid: str) -> RottedLinks:
-    return db.session.execute(db.select(RottedLinks).filter_by(id=uuid)).scalars().first()
+    return db.session.execute(db.select(RottedLinks).filter_by(id=uuid)).scalar_one_or_none()
 
 
-def __update(data: RottedLinks) -> Literal[True]:
-    db.session.query(RottedLinks).filter_by(id=data.id).update(
-        {"times_failed": data.times_failed + 1}, synchronize_session="fetch"
-    )
+def __update(rl: RottedLinks) -> Literal[True]:
+    """Update the rotted log link with this instance."""
+    data = rl.as_dict()
+    del data["id"]
+    data["times_failed"] += 1
+    rl.update_with(data)
     db.session.commit()
     return True
 

diff --git a/src/core/database/weblink.py b/src/core/database/weblink.py
@@ -1,7 +1,7 @@
 import uuid
 from collections import OrderedDict
-from datetime import UTC, datetime, timezone
-from typing import Any, Optional
+from datetime import UTC, datetime
+from typing import Any
 
 from markupsafe import Markup
 from sqlalchemy import func
@@ -55,7 +55,7 @@ def exists(uuid: str) -> bool:
 
 def get(uuid: str) -> WebLink | None:
     """Get a single weblink."""
-    return db.session.execute(db.select(WebLink).filter_by(id=uuid)).scalars().first()
+    return db.session.execute(db.select(WebLink).filter_by(id=uuid)).scalar_one_or_none()
 
 
 def get_all(include_rotted: bool = False, **kwargs: Any) -> list[WebLink]:
@@ -83,16 +83,14 @@ def get_all(include_rotted: bool = False, **kwargs: Any) -> list[WebLink]:
 
 def update(data: OrderedDict) -> bool:
     """Update a weblink."""
-    if not exists(data["id"]):
+    wb_id = data.pop("id")
+    if (wb := get(wb_id)) is None:
         return False
 
-    db.session.query(WebLink).filter_by(id=data["id"]).update(
-        {k: Markup(v).striptags() for k, v in data.items() if k != "id"},
-        synchronize_session="fetch",
-    )
+    wb.update_with(data)
     db.session.commit()
     logger.info({
-        "id": data["id"],
+        "id": wb_id,
         "url": "N/A",
         "message": f"Link has been updated with the following info: `{data}`",
     })