42 lines
1.1 KiB
Python
42 lines
1.1 KiB
Python
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")
|