Bump pandas from 2.2.3 to 2.3.3 (ansible#252)

Bumps [pandas](https://github.com/pandas-dev/pandas) from 2.2.3 to 2.3.3.
- [Release notes](https://github.com/pandas-dev/pandas/releases)
- [Commits](pandas-dev/pandas@v2.2.3...v2.3.3)

---
updated-dependencies:
- dependency-name: pandas
  dependency-version: 2.3.3
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Author: dependabot[bot]

metrics_utility/automation_controller_billing/collector.py

@@ -1,27 +1,20 @@
-import contextlib
 import json
 import os
 
+from contextlib import contextmanager
+
 from django.conf import settings
 from django.core.serializers.json import DjangoJSONEncoder
 from django.db import connection
 
 import metrics_utility.base as base
 
+from metrics_utility.automation_controller_billing.db_lock_helpers import advisory_lock
 from metrics_utility.automation_controller_billing.helpers import get_last_entries_from_db
 from metrics_utility.automation_controller_billing.package.factory import Factory as PackageFactory
 from metrics_utility.logger import logger
 
 
-# work around https://github.com/ansible/awx/pull/15676
-try:
-    # 2.4, early 2.5
-    from awx.main.utils.pglock import advisory_lock
-except ImportError:
-    # later 2.5, 2.6
-    from ansible_base.lib.utils.db import advisory_lock
-
-
 class Collector(base.Collector):
     def __init__(self, collection_type=base.Collector.SCHEDULED_COLLECTION, collector_module=None, ship_target=None, billing_provider_params=None):
         if collector_module is None:
@@ -106,7 +99,7 @@ def registered_collectors(cls, module=None):
 
         return base.Collector.registered_collectors(collectors)
 
-    @contextlib.contextmanager
+    @contextmanager
     def _pg_advisory_lock(self, key, wait=False):
         """Use awx specific implementation to pass tests with sqlite3"""
         with advisory_lock(key, wait=wait) as lock:

metrics_utility/automation_controller_billing/db_lock_helpers.py

@@ -0,0 +1,87 @@
+from contextlib import contextmanager
+from zlib import crc32
+
+from django.db import DEFAULT_DB_ALIAS, connection, connections
+
+from metrics_utility.logger import logger
+
+
+@contextmanager
+def django_pglocks_advisory_lock(key, shared=False, wait=False, db_alias=DEFAULT_DB_ALIAS):
+    function_name = 'pg_'
+    release_function_name = 'pg_advisory_unlock'
+    if not wait:
+        function_name += 'try_'
+    function_name += 'advisory_lock'
+    if shared:
+        function_name += '_shared'
+        release_function_name += '_shared'
+    tuple_format = False
+    if isinstance(key, (list, tuple)):
+        if len(key) != 2:
+            raise ValueError('Tuples and lists as lock IDs must have exactly two entries.')
+        if not isinstance(key[0], int) or not isinstance(key[1], int):
+            raise ValueError('Both members of a tuple/list lock ID must be integers')
+        tuple_format = True
+    elif isinstance(key, str):
+        pos = crc32(key.encode('utf-8'))
+        key = (2**31 - 1) & pos
+        if pos & 2**31:
+            key -= 2**31
+    elif not isinstance(key, int):
+        raise ValueError('Cannot use %s as a lock id' % key)
+    if tuple_format:
+        base = 'SELECT %s(%d, %d)'
+        params = (key[0], key[1])
+    else:
+        base = 'SELECT %s(%d)'
+        params = (key,)
+    acquire_params = (function_name,) + params
+    command = base % acquire_params
+
+    with connections[db_alias].cursor() as cursor:
+        cursor.execute(command)
+
+        if not wait:
+            acquired = cursor.fetchone()[0]
+        else:
+            acquired = True
+        try:
+            yield acquired
+        except Exception as e:
+            if not acquired and not wait:
+                raise e
+        finally:
+            if acquired:
+                release_params = (release_function_name,) + params
+
+                command = base % release_params
+                cursor.execute(command)
+
+
+@contextmanager
+def advisory_lock(*args, lock_session_timeout_milliseconds=0, **kwargs):
+    if connection.vendor == 'postgresql':
+        cur = None
+        idle_in_transaction_session_timeout = None
+        idle_session_timeout = None
+        try:
+            if lock_session_timeout_milliseconds > 0:
+                """Set the idle session timeout for the current transaction"""
+                with connection.cursor() as cur:
+                    idle_in_transaction_session_timeout = cur.execute('SHOW idle_in_transaction_session_timeout').fetchone()[0]
+                    idle_session_timeout = cur.execute('SHOW idle_session_timeout').fetchone()[0]
+                    cur.execute('SET idle_in_transaction_session_timeout = %s', (lock_session_timeout_milliseconds,))
+                    cur.execute('SET idle_session_timeout = %s', (lock_session_timeout_milliseconds,))
+            with django_pglocks_advisory_lock(*args, **kwargs) as internal_lock:
+                yield internal_lock
+                if lock_session_timeout_milliseconds > 0:
+                    with connection.cursor() as cur:
+                        cur.execute('SET idle_in_transaction_session_timeout = %s', (idle_in_transaction_session_timeout,))
+                        cur.execute('SET idle_session_timeout = %s', (idle_session_timeout,))
+        except Exception as e:
+            logger.error(f'Error setting idle session timeout: {e}')
+    elif connection.vendor == 'sqlite':
+        yield True
+    else:
+        raise RuntimeError(f'Advisory lock not implemented for database type {connection.vendor}')

metrics_utility/base/collector.py

@@ -1,5 +1,4 @@
 import contextlib
-import hashlib
 import inspect
 import logging
 import os
@@ -9,8 +8,10 @@
 
 from abc import abstractmethod
 
+from django.db import DEFAULT_DB_ALIAS
 from django.utils.timezone import now, timedelta
 
+from metrics_utility.automation_controller_billing.db_lock_helpers import advisory_lock
 from metrics_utility.logger import logger
 
 from .collection import Collection
@@ -332,29 +333,51 @@ def _add_collection_to_package(self, collection):
         if collection.ship_immediately():
             self._process_package(package)
 
+    @abstractmethod
     @contextlib.contextmanager
-    def _pg_advisory_lock(self, key, wait=False):
-        """Postgres db lock"""
-        connection = self.db_connection()
-
-        if connection is None:
-            yield True
-        else:
-            # Build 64-bit integer out of the resource id
-            resource_key = int(hashlib.sha512(key.encode()).hexdigest(), 16) % 2**63
-
-            cursor = connection.cursor()
-
-            try:
-                if wait:
-                    cursor.execute('SELECT pg_advisory_lock(%s);', (resource_key,))
-                else:
-                    cursor.execute('SELECT pg_try_advisory_lock(%s);', (resource_key,))
-                acquired = cursor.fetchall()[0][0]
-                yield acquired
-            finally:
-                cursor.execute('SELECT pg_advisory_unlock(%s);', (resource_key,))
-                cursor.close()
+    def _pg_advisory_lock(self, key, shared=False, wait=False, db_alias=DEFAULT_DB_ALIAS):
+        """PostgreSQL advisory lock context manager.
+
+        Subclasses must implement this method to provide database advisory locking.
+        The implementation should:
+
+        Args:
+            key (str): Lock identifier. Can be a string which will be converted to an integer
+                      using a hash function (CRC32, SHA512, etc.), or an integer, or a tuple
+                      of two integers for PostgreSQL's two-parameter lock functions.
+            shared (bool): If True, acquire a shared lock (pg_advisory_lock_shared).
+                          If False, acquire an exclusive lock (pg_advisory_lock).
+                          Default is False (exclusive lock).
+            wait (bool): If True, block until lock is acquired (pg_advisory_lock).
+                        If False, return immediately (pg_try_advisory_lock).
+                        When wait=False and lock cannot be acquired, should raise an exception.
+                        Default is False (non-blocking).
+            db_alias (str): Django database alias to use for the connection.
+                           Default is DEFAULT_DB_ALIAS (usually 'default').
+
+        Yields:
+            bool: True if lock was acquired, False if not acquired (only when wait=True).
+                  When wait=False, should either yield True or raise an exception.
+
+        The implementation should:
+        - Support PostgreSQL advisory locks using pg_advisory_lock/pg_try_advisory_lock
+        - Handle lock release automatically in finally block using pg_advisory_unlock
+        - Support both shared and exclusive locks if needed
+        - Handle different database vendors (PostgreSQL, SQLite, etc.)
+        - Convert string keys to appropriate integer format for PostgreSQL
+        - Raise exceptions when wait=False and lock cannot be acquired
+
+        Example PostgreSQL lock functions used:
+        - pg_advisory_lock(key) - blocking exclusive lock
+        - pg_try_advisory_lock(key) - non-blocking exclusive lock
+        - pg_advisory_lock_shared(key) - blocking shared lock
+        - pg_try_advisory_lock_shared(key) - non-blocking shared lock
+        - pg_advisory_unlock(key) - release exclusive lock
+        - pg_advisory_unlock_shared(key) - release shared lock
+        """
+        with advisory_lock(key, shared=shared, wait=wait, db_alias=db_alias) as lock:
+            yield lock
+        pass
 
     def _process_packages(self):
         for group, packages in self.packages.items():

metrics_utility/test/base/classes/analytics_collector.py

@@ -1,5 +1,9 @@
+from contextlib import contextmanager
+
 from base.classes.package import Package
+from django.db import DEFAULT_DB_ALIAS
 
+from metrics_utility.automation_controller_billing.db_lock_helpers import advisory_lock
 from metrics_utility.base import Collector
 
 
@@ -12,6 +16,11 @@ def db_connection():
     def _package_class():
         return Package
 
+    @contextmanager
+    def _pg_advisory_lock(self, key, shared=False, wait=False, db_alias=DEFAULT_DB_ALIAS):
+        with advisory_lock(key, shared=shared, wait=wait, db_alias=db_alias) as lock:
+            yield lock
+
     def _is_shipping_configured(self):
         return False