code: def with_row_locks(query: Query, session: Session, **kwargs) -> Query:
      """
      Apply with_for_update to an SQLAlchemy query, if row level locking is in use.
  
      :param query: An SQLAlchemy Query object
      :param session: ORM Session
      :param kwargs: Extra kwargs to pass to with_for_update (of, nowait, skip_locked, etc)
      :return: updated query
      """
      dialect = session.bind.dialect
  
      # Don't use row level locks if the MySQL dialect (Mariadb & MySQL < 8) does not support it.
      if USE_ROW_LEVEL_LOCKING and (
          dialect.name != "mysql" or dialect.supports_for_update_of
      ):
          return query.with_for_update(**kwargs)
      else:
          return query
  
imports: from __future__ import annotations
  import contextlib
  import copy
  import datetime
  import json
  import logging
  from typing import TYPE_CHECKING, Any, Generator, Iterable, overload
  from dateutil import relativedelta
  from sqlalchemy import TIMESTAMP, PickleType, event, nullsfirst, tuple_
  from sqlalchemy.dialects import mysql
  from sqlalchemy.types import JSON, Text, TypeDecorator
  from airflow.configuration import conf
  from airflow.serialization.enums import Encoding
  from airflow.utils.timezone import make_naive, utc
label_desc: Use image node_id 6 for calling a global function with example usage: with_row_locks(query, session) and returns: query, query