小程序后端补充数据库兼容层与启动适配

mini-program/app/db_compat.py

@@ -0,0 +1,41 @@
+from collections.abc import Sequence
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+
+def get_db_dialect_name(db: Session) -> str:
+    bind = db.get_bind()
+    if bind is None:
+        raise RuntimeError("Database session is not bound to an engine")
+    return bind.dialect.name
+
+
+def current_timestamp_sql(db: Session) -> str:
+    if get_db_dialect_name(db) == "mysql":
+        return "CURRENT_TIMESTAMP(3)"
+    return "CURRENT_TIMESTAMP"
+
+
+def select_for_update_clause(db: Session) -> str:
+    if get_db_dialect_name(db) == "mysql":
+        return " FOR UPDATE"
+    return ""
+
+
+def inserted_primary_key(result: Any) -> int:
+    lastrowid = getattr(result, "lastrowid", None)
+    if lastrowid is not None:
+        return int(lastrowid)
+
+    try:
+        inserted_primary_key = result.inserted_primary_key
+    except Exception:
+        inserted_primary_key = None
+
+    if isinstance(inserted_primary_key, Sequence) and inserted_primary_key:
+        primary_key = inserted_primary_key[0]
+        if primary_key is not None:
+            return int(primary_key)
+
+    raise RuntimeError("Could not determine inserted primary key")