35 lines
978 B
Python
35 lines
978 B
Python
from typing import Any
|
|
|
|
from sqlalchemy.orm import DeclarativeBase, Session
|
|
|
|
|
|
def _query_existing[DbModelT: DeclarativeBase](
|
|
session: Session, DbModel: type[DbModelT], query_parameters: dict[str, Any]
|
|
) -> DbModelT | None:
|
|
query = session.query(DbModel).filter_by(**query_parameters)
|
|
return query.first()
|
|
|
|
|
|
def upsert[DbModelT: DeclarativeBase](
|
|
session: Session,
|
|
DbModel: type[DbModelT],
|
|
parameters_existing: dict[str, Any],
|
|
parameters_new: dict[str, Any],
|
|
do_update: bool = False,
|
|
do_commit: bool = True,
|
|
) -> DbModelT:
|
|
existing_db_row = _query_existing(session, DbModel, parameters_existing)
|
|
|
|
if existing_db_row:
|
|
if do_update:
|
|
for key, value in parameters_new.items():
|
|
setattr(existing_db_row, key, value)
|
|
return existing_db_row
|
|
|
|
new_db_row = DbModel(**parameters_new)
|
|
session.add(new_db_row)
|
|
session.flush()
|
|
if do_commit:
|
|
session.commit()
|
|
return new_db_row
|