chore: update log adapter to support session-scoped api method logging

Updates `log_adapter` to optionally store API method names in a `Session` instance instead of the global list. This improves label accuracy when running tests in parallel.

- Updates `Session` to initialize `_api_methods` list and lock.
- Updates `log_adapter.add_api_method` and `get_and_reset_api_methods` to handle session-scoped logging.
- Updates `log_adapter.method_logger` and `property_logger` to identify the session from arguments.
- Propagates `session` through `start_query_with_client` and its callers to ensure labels are correctly associated with the session.
- Refactored `add_api_method` to be more DRY.
- Fixes method signatures in `create_temp_table` and `create_temp_view`.

Author: google-labs-jules[bot]

bigframes/core/log_adapter.py

@@ -309,10 +309,19 @@ def _find_session(*args, **kwargs):
     from bigframes.session import Session
 
     if args and isinstance(args[0], Session):
-        return args[0]
+        # In unit tests, we might be working with a mock Session object that
+        # passes isinstance but doesn't have the instance attributes set in
+        # __init__.
+        session = args[0]
+        if hasattr(session, "_api_methods_lock") and hasattr(session, "_api_methods"):
+            return session
 
     session = kwargs.get("session")
     if session is not None and isinstance(session, Session):
-        return session
+        # In unit tests, we might be working with a mock Session object that
+        # passes isinstance but doesn't have the instance attributes set in
+        # __init__.
+        if hasattr(session, "_api_methods_lock") and hasattr(session, "_api_methods"):
+            return session
 
     return None

bigframes/session/__init__.py

@@ -23,6 +23,7 @@
 import logging
 import os
 import secrets
+import threading
 import typing
 from typing import (
     Any,

bigframes/session/_io/bigquery/__init__.py

@@ -126,6 +126,7 @@ def create_temp_table(
     schema: Optional[Iterable[bigquery.SchemaField]] = None,
     cluster_columns: Optional[list[str]] = None,
     kms_key: Optional[str] = None,
+    session=None,
 ) -> str:
     """Create an empty table with an expiration in the desired session.
 
@@ -153,6 +154,7 @@ def create_temp_view(
     *,
     expiration: datetime.datetime,
     sql: str,
+    session=None,
 ) -> str:
     """Create an empty table with an expiration in the desired session.