add banban service code
This commit is contained in:
22
talkingq-url/banban/db_compat.py
Normal file
22
talkingq-url/banban/db_compat.py
Normal file
@@ -0,0 +1,22 @@
|
||||
from collections.abc import Sequence
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
|
||||
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")
|
||||
Reference in New Issue
Block a user