Update stats module to use SqlAlchemy 2.x syntax

Author: varmar05

server/mergin/stats/models.py

@@ -2,11 +2,12 @@
 #
 # SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-MerginMaps-Commercial
 
+import uuid
 from dataclasses import dataclass
 from typing import Optional
-import uuid
-from sqlalchemy.dialects.postgresql import UUID, JSONB
-from datetime import datetime, timezone
+from datetime import datetime
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
 
 from ..app import db
 
@@ -30,8 +31,8 @@ class ServerCallhomeData:
 class MerginInfo(db.Model):
     """Information about deployment"""
 
-    service_id = db.Column(UUID(as_uuid=True), primary_key=True)
-    last_reported = db.Column(db.DateTime)
+    service_id: Mapped[uuid.UUID] = mapped_column(primary_key=True)
+    last_reported: Mapped[Optional[datetime]]
 
     def __init__(self, service_id: str = None):
         if service_id:
@@ -43,9 +44,7 @@ def __init__(self, service_id: str = None):
 class MerginStatistics(db.Model):
     """Information about deployment"""
 
-    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
-    created_at = db.Column(
-        db.DateTime, index=True, nullable=False, default=datetime.utcnow
-    )
+    id: Mapped[int] = mapped_column(primary_key=True)
+    created_at: Mapped[datetime] = mapped_column(index=True, default=datetime.utcnow)
     # data with statistics
-    data = db.Column(JSONB, nullable=False)
+    data: Mapped[dict] = mapped_column(JSONB)

server/mergin/stats/tasks.py

@@ -10,6 +10,7 @@
 from flask import current_app
 from sqlalchemy.sql.operators import is_
 from sqlalchemy import inspect
+from sqlalchemy.sql import select, func
 
 from .models import MerginInfo, MerginStatistics, ServerCallhomeData
 from ..celery import celery
@@ -22,21 +23,25 @@ def get_callhome_data(info: MerginInfo | None = None) -> ServerCallhomeData:
     """
     Get data about server to send to callhome service
     """
-    last_change_item = (
-        db.session.query(Project.updated).order_by(Project.updated.desc()).first()
-    )
+    last_change = db.session.scalars(
+        select(Project.updated).order_by(Project.updated.desc())
+    ).first()
     service_uuid = str(info.service_id) if info else None
     data = ServerCallhomeData(
         service_uuid=service_uuid,
         url=current_app.config["MERGIN_BASE_URL"],
         contact_email=current_app.config["CONTACT_EMAIL"],
         licence=current_app.config["SERVER_TYPE"],
-        projects_count=Project.query.filter(Project.removed_at.is_(None)).count(),
-        users_count=User.query.filter(
-            is_(User.username.ilike("deleted_%"), False)
-        ).count(),
+        projects_count=db.session.scalar(
+            select(func.count(Project.id)).where(is_(Project.removed_at, None))
+        ),
+        users_count=db.session.scalar(
+            select(func.count(User.id)).where(
+                is_(User.username.ilike("deleted_%"), False)
+            )
+        ),
         workspaces_count=current_app.ws_handler.workspace_count(),
-        last_change=str(last_change_item.updated) + "Z" if last_change_item else "",
+        last_change=str(last_change) + "Z" if last_change else "",
         server_version=current_app.config["VERSION"],
         monthly_contributors=current_app.ws_handler.monthly_contributors_count(),
         editors=current_app.ws_handler.server_editors_count(),
@@ -48,7 +53,7 @@ def get_callhome_data(info: MerginInfo | None = None) -> ServerCallhomeData:
 @celery.task(ignore_result=True)
 def save_statistics():
     """Save statistics about usage."""
-    info = MerginInfo.query.first()
+    info = db.session.execute(select(MerginInfo)).scalar_one_or_none()
     data = get_callhome_data(info)
     stat = MerginStatistics(data=data)
     db.session.add(stat)
@@ -77,7 +82,7 @@ def send_statistics():
         logging.warning("Database not initialized")
         return
 
-    info = MerginInfo.query.first()
+    info = db.session.execute(select(MerginInfo)).scalar_one_or_none()
     if not info:
         # create new info with random service id
         service_id = current_app.config.get("SERVICE_ID", None)