From eda5baa79e274fab8430348be803c70014c49cd6 Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Thu, 9 Jul 2026 17:53:30 -0700 Subject: [PATCH 1/6] feat(functions): Add FunctionScope and update task_queue to support kit scopes Added FunctionScope class, updated task_queue signature, and refactored TaskQueue to implement rollback-safe recursive fallback targeting kit queue when hitting a 404 response. --- firebase_admin/functions.py | 206 +++++++++++++++++++++++++----- tests/test_functions.py | 242 +++++++++++++++++++++++++++++++++++- 2 files changed, 411 insertions(+), 37 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index 66ba700b..fadc0ea2 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -19,9 +19,10 @@ from urllib import parse import re import os +import warnings import json from base64 import b64encode -from typing import Any, Optional, Dict +from typing import Any, Optional, Dict, Union from dataclasses import dataclass from google.auth.compute_engine import Credentials as ComputeEngineCredentials @@ -38,12 +39,53 @@ _FUNCTIONS_ATTRIBUTE = '_functions' __all__ = [ + 'FunctionScope', 'TaskOptions', 'task_queue', ] +class FunctionScope: + """Defines the scope of a function in a task queue.""" + + def __init__(self, type_: str, instance_id: Optional[str] = None): + if type_ not in ( + 'current', 'global', 'extension', 'kit', 'extension_or_kit'): + # "kit" is not released yet and "extension_or_kit" is an implementation detail, + # so they are not included in the public list of options. + raise ValueError( + 'type_ must be one of "current", "global", or "extension".' + ) + if type_ in ('extension', 'kit', 'extension_or_kit'): + if not isinstance(instance_id, str): + raise ValueError('instance_id must be a string.') + if instance_id == '': + raise ValueError('instance_id must be a non-empty string.') + else: + if instance_id is not None: + raise ValueError( + f'instance_id must not be set for type "{type_}".' + ) + self.type = type_ + self.instance_id = instance_id + + @classmethod + def current(cls): + """Creates a FunctionScope targeting the current extension or kit context.""" + return cls("current") + + @classmethod + def global_scope(cls): + """Creates a FunctionScope targeting a global function.""" + return cls("global") + + @classmethod + def extension(cls, extension_id: str): + """Creates a FunctionScope targeting a specific extension instance.""" + return cls("extension", instance_id=extension_id) + + _CLOUD_TASKS_API_RESOURCE_PATH = \ 'projects/{project_id}/locations/{location_id}/queues/{resource_id}/tasks' _CLOUD_TASKS_API_URL_FORMAT = \ @@ -77,8 +119,9 @@ def _get_functions_service(app) -> _FunctionsService: def task_queue( function_name: str, - extension_id: Optional[str] = None, - app: Optional[App] = None + scope: Optional[Union[str, FunctionScope]] = None, + app: Optional[App] = None, + **kwargs ) -> TaskQueue: """Creates a reference to a TaskQueue for a given function name. @@ -96,8 +139,9 @@ def task_queue( Args: function_name: Name of the function. - extension_id: Firebase extension ID (optional). + scope: FunctionScope configuration or Firebase extension ID (optional). app: An App instance (optional). + **kwargs: Backwards-compatible parameters (e.g., extension_id). Returns: TaskQueue: A TaskQueue instance. @@ -105,7 +149,21 @@ def task_queue( Raises: ValueError: If the input arguments are invalid. """ - return _get_functions_service(app).task_queue(function_name, extension_id) + extension_id = kwargs.pop('extension_id', None) + if kwargs: + raise TypeError(f"task_queue() got an unexpected keyword argument '{next(iter(kwargs))}'") + + if extension_id is not None: + if scope is not None: + raise ValueError('Cannot set both scope and extension_id.') + warnings.warn( + 'extension_id is deprecated. Use scope instead.', + DeprecationWarning, + stacklevel=2 + ) + scope = extension_id + + return _get_functions_service(app).task_queue(function_name, scope) class _FunctionsService: """Service class that implements Firebase Functions functionality.""" @@ -125,10 +183,14 @@ def __init__(self, app: App): self._http_client = _http_client.JsonHttpClient(credential=self._credential) - def task_queue(self, function_name: str, extension_id: Optional[str] = None) -> TaskQueue: + def task_queue( + self, + function_name: str, + scope: Optional[Union[str, FunctionScope]] = None + ) -> TaskQueue: """Creates a TaskQueue instance.""" return TaskQueue( - function_name, extension_id, self._project_id, self._credential, self._http_client, + function_name, scope, self._project_id, self._credential, self._http_client, self._emulator_host) @classmethod @@ -142,7 +204,7 @@ class TaskQueue: def __init__( self, function_name: str, - extension_id: Optional[str], + scope: Optional[Union[str, FunctionScope]], project_id, credential, http_client, @@ -157,7 +219,18 @@ def __init__( self._http_client = http_client self._emulator_host = emulator_host self._function_name = function_name - self._extension_id = extension_id + + # Determine the scope + if scope is None: + self._scope = FunctionScope.current() + elif isinstance(scope, str): + _Validators.check_non_empty_string('scope', scope) + self._scope = FunctionScope('extension_or_kit', scope) + elif isinstance(scope, FunctionScope): + self._scope = scope + else: + raise ValueError('scope must be a string or a FunctionScope object.') + # Parse resources from function_name self._resource = self._parse_resource_name(self._function_name, 'functions') @@ -165,11 +238,59 @@ def __init__( self._resource.project_id = self._resource.project_id or self._project_id self._resource.location_id = self._resource.location_id or _DEFAULT_LOCATION _Validators.check_non_empty_string('resource.resource_id', self._resource.resource_id) - # Validate extension_id if provided and edit resources depending - if self._extension_id is not None: - _Validators.check_non_empty_string('extension_id', self._extension_id) - self._resource.resource_id = f'ext-{self._extension_id}-{self._resource.resource_id}' + def _resolve_resource(self, scope: FunctionScope) -> tuple[Resource, Optional[str]]: + """Resolves the Resource object and extension/kit ID based on the scope.""" + resource_id = self._resource.resource_id + extension_or_kit_id = None + + if scope.type == 'current': + kit_instance_id = os.environ.get('KIT_INSTANCE_ID') + if kit_instance_id: + resource_id = f'kit-{kit_instance_id}-{resource_id}' + extension_or_kit_id = kit_instance_id + elif scope.type == 'global': + pass + elif scope.type in ('extension', 'extension_or_kit'): + resource_id = f'ext-{scope.instance_id}-{resource_id}' + extension_or_kit_id = scope.instance_id + elif scope.type == 'kit': + resource_id = f'kit-{scope.instance_id}-{resource_id}' + extension_or_kit_id = scope.instance_id + + resolved = Resource( + resource_id=resource_id, + project_id=self._resource.project_id, + location_id=self._resource.location_id + ) + return resolved, extension_or_kit_id + + def _log_fallback_warning(self, function_name: str, instance: str) -> None: + """Logs a warning when falling back from extension to kit.""" + warnings.warn( + f"Targeting kit {instance} with the legacy extensions API, " + "which has performance implications. Please change the call " + f"task_queue('{function_name}', '{instance}') to " + f"task_queue('{function_name}', FunctionScope('kit', '{instance}'))", + UserWarning, + stacklevel=3 + ) + + def _parse_enqueue_response(self, resp: dict, resolved_resource: Resource) -> str: + """Parses the enqueue response to extract the task ID.""" + if self._is_emulated(): + # Emulator returns a response with format {task: {name: }} + # The task name also has an extra '/' at the start compared to prod + task_info = resp.get('task') or {} + task_name = task_info.get('name') + if task_name: + task_name = task_name[1:] + else: + # Production returns a response with format {name: } + task_name = resp.get('name') + task_resource = self._parse_resource_name( + task_name, f'queues/{resolved_resource.resource_id}/tasks') + return task_resource.resource_id def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str: """Creates a task and adds it to the queue. Tasks cannot be updated after creation. @@ -188,10 +309,11 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str: Returns: str: The ID of the task relative to this queue. """ - task = self._validate_task_options(task_data, self._resource, opts) - emulator_url = self._get_emulator_url(self._resource) - service_url = emulator_url or self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT) - task_payload = self._update_task_payload(task, self._resource, self._extension_id) + resolved_resource, ext_or_kit_id = self._resolve_resource(self._scope) + task = self._validate_task_options(task_data, resolved_resource, opts) + service_url = self._get_emulator_url(resolved_resource) or self._get_url( + resolved_resource, _CLOUD_TASKS_API_URL_FORMAT) + task_payload = self._update_task_payload(task, resolved_resource, ext_or_kit_id) try: resp = self._http_client.body( 'post', @@ -199,21 +321,22 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str: headers=_FUNCTIONS_HEADERS, json={'task': task_payload.to_api_dict()} ) - if self._is_emulated(): - # Emulator returns a response with format {task: {name: }} - # The task name also has an extra '/' at the start compared to prod - task_info = resp.get('task') or {} - task_name = task_info.get('name') - if task_name: - task_name = task_name[1:] - else: - # Production returns a response with format {name: } - task_name = resp.get('name') - task_resource = \ - self._parse_resource_name(task_name, f'queues/{self._resource.resource_id}/tasks') - return task_resource.resource_id + return self._parse_enqueue_response(resp, resolved_resource) except requests.exceptions.RequestException as error: - raise _FunctionsService.handle_functions_error(error) + is_404 = error.response is not None and error.response.status_code == 404 + if self._scope.type != 'extension_or_kit' or not is_404: + raise _FunctionsService.handle_functions_error(error) + + # Upgrade scope to kit and retry recursively, rolling back if it throws + old_scope = self._scope + self._scope = FunctionScope('kit', self._scope.instance_id) + try: + resp = self.enqueue(task_data, opts) + self._log_fallback_warning(self._function_name, self._scope.instance_id) + return resp + except Exception: + self._scope = old_scope + raise def delete(self, task_id: str) -> None: """Deletes an enqueued task if it has not yet started. @@ -229,11 +352,14 @@ def delete(self, task_id: str) -> None: ValueError: If the input arguments are invalid. """ _Validators.check_non_empty_string('task_id', task_id) - emulator_url = self._get_emulator_url(self._resource) + + resolved_resource, _ = self._resolve_resource(self._scope) + emulator_url = self._get_emulator_url(resolved_resource) if emulator_url: service_url = emulator_url + f'/{task_id}' else: - service_url = self._get_url(self._resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}') + service_url = self._get_url( + resolved_resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}') try: self._http_client.body( 'delete', @@ -241,7 +367,19 @@ def delete(self, task_id: str) -> None: headers=_FUNCTIONS_HEADERS, ) except requests.exceptions.RequestException as error: - raise _FunctionsService.handle_functions_error(error) + is_404 = error.response is not None and error.response.status_code == 404 + if not is_404: + raise _FunctionsService.handle_functions_error(error) + + if self._scope.type == 'extension_or_kit': + old_scope = self._scope + self._scope = FunctionScope('kit', self._scope.instance_id) + try: + self.delete(task_id) + self._log_fallback_warning(self._function_name, self._scope.instance_id) + except Exception: + self._scope = old_scope + raise def _parse_resource_name(self, resource_name: str, resource_id_key: str) -> Resource: diff --git a/tests/test_functions.py b/tests/test_functions.py index 0f766767..1682ca5a 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -17,6 +17,7 @@ from datetime import datetime, timedelta, timezone import json import time +import warnings import pytest import firebase_admin @@ -97,19 +98,21 @@ def test_task_queue_invalid_function_name_error(self, function_name): def test_task_queue_extension_id(self): queue = functions.task_queue("test-function-name", "test-extension-id") - assert queue._resource.resource_id == 'ext-test-extension-id-test-function-name' + assert queue._scope.type == 'extension_or_kit' + assert queue._scope.instance_id == 'test-extension-id' + assert queue._resource.resource_id == 'test-function-name' assert queue._resource.project_id == 'test-project' assert queue._resource.location_id == 'us-central1' def test_task_queue_empty_extension_id_error(self): with pytest.raises(ValueError) as excinfo: functions.task_queue('test-function-name', '') - assert str(excinfo.value) == 'extension_id "" must be a non-empty string.' + assert str(excinfo.value) == 'scope "" must be a non-empty string.' def test_task_queue_non_string_extension_id_error(self): with pytest.raises(ValueError) as excinfo: functions.task_queue('test-function-name', 1234) - assert str(excinfo.value) == 'extension_id "1234" must be a string.' + assert str(excinfo.value) == 'scope must be a string or a FunctionScope object.' def test_task_enqueue(self): @@ -262,6 +265,206 @@ def test_get_emulator_url_invalid_format(self, monkeypatch): functions.task_queue('test-function-name', app=app) assert 'Invalid CLOUD_TASKS_EMULATOR_HOST' in str(excinfo.value) + def test_task_queue_with_scope_parameter(self): + scope = functions.FunctionScope.extension('my-ext') + queue = functions.task_queue('test-function-name', scope=scope) + assert queue._scope == scope + + def test_task_queue_deprecation_warning(self): + with pytest.warns(DeprecationWarning) as record: + queue = functions.task_queue('test-function-name', extension_id='my-ext') + assert queue._scope.type == 'extension_or_kit' + assert queue._scope.instance_id == 'my-ext' + assert len(record) == 1 + assert 'extension_id is deprecated. Use scope instead.' in str(record[0].message) + + def test_enqueue_current_scope_default(self): + _, recorder = self._instrument_functions_service() + queue = functions.task_queue('test-function-name') + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + 'queues/test-function-name/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_current_scope_env_vars_ext(self, monkeypatch): + # Python SDK does not handle EXT_INSTANCE_ID since no extensions use Python + monkeypatch.setenv('EXT_INSTANCE_ID', 'my-ext') + _, recorder = self._instrument_functions_service() + queue = functions.task_queue('test-function-name') + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + 'queues/test-function-name/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_current_scope_env_vars_kit(self, monkeypatch): + monkeypatch.setenv('KIT_INSTANCE_ID', 'my-kit') + expected_queue = 'kit-my-kit-test-function-name' + response_payload = json.dumps({ + 'name': ( + f'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks/test-task-id' + ) + }) + _, recorder = self._instrument_functions_service(payload=response_payload) + queue = functions.task_queue('test-function-name') + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_global_scope(self): + _, recorder = self._instrument_functions_service() + queue = functions.task_queue( + 'test-function-name', scope=functions.FunctionScope.global_scope()) + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + 'queues/test-function-name/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_extension_scope(self): + expected_queue = 'ext-my-ext-test-function-name' + response_payload = json.dumps({ + 'name': ( + f'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks/test-task-id' + ) + }) + _, recorder = self._instrument_functions_service(payload=response_payload) + queue = functions.task_queue( + 'test-function-name', scope=functions.FunctionScope.extension('my-ext')) + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_kit_scope(self): + expected_queue = 'kit-my-kit-test-function-name' + response_payload = json.dumps({ + 'name': ( + f'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks/test-task-id' + ) + }) + _, recorder = self._instrument_functions_service(payload=response_payload) + queue = functions.task_queue( + 'test-function-name', scope=functions.FunctionScope('kit', 'my-kit')) + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks' + ) + assert recorder[0].url == expected_url + + def test_enqueue_fallback_flow(self): + functions_service = functions._get_functions_service(firebase_admin.get_app()) + recorder = [] + task_name = ( + 'projects/test-project/locations/us-central1/queues/' + 'kit-my-instance-test-function-name/tasks/task-123' + ) + adapter = testutils.MockMultiRequestAdapter( + [json.dumps({'error': {'status': 'NOT_FOUND'}}), + json.dumps({'name': task_name})], + [404, 200], + recorder + ) + functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + + queue = functions.task_queue('test-function-name', 'my-instance') + with pytest.warns(UserWarning) as record: + task_id = queue.enqueue(_DEFAULT_DATA) + assert task_id == 'task-123' + assert queue._scope.type == 'kit' + assert len(record) == 1 + assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) + assert len(recorder) == 2 + assert 'ext-my-instance-test-function-name' in recorder[0].url + assert 'kit-my-instance-test-function-name' in recorder[1].url + + def test_delete_fallback_flow(self): + functions_service = functions._get_functions_service(firebase_admin.get_app()) + recorder = [] + adapter = testutils.MockMultiRequestAdapter( + [json.dumps({'error': {'status': 'NOT_FOUND'}}), '{}'], + [404, 200], + recorder + ) + functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + + queue = functions.task_queue('test-function-name', 'my-instance') + with pytest.warns(UserWarning) as record: + queue.delete('task-123') + assert queue._scope.type == 'kit' + assert len(record) == 1 + assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) + assert len(recorder) == 2 + assert 'ext-my-instance-test-function-name/tasks/task-123' in recorder[0].url + assert 'kit-my-instance-test-function-name/tasks/task-123' in recorder[1].url + + def test_delete_ignoring_404(self): + _, recorder = self._instrument_functions_service( + status=404, payload='{"error": {"status": "NOT_FOUND"}}') + queue = functions.task_queue( + 'test-function-name', scope=functions.FunctionScope.global_scope()) + queue.delete('task-123') + assert len(recorder) == 1 + + def test_enqueue_fallback_failure_reverts_scope(self): + functions_service = functions._get_functions_service(firebase_admin.get_app()) + recorder = [] + adapter = testutils.MockMultiRequestAdapter( + [json.dumps({'error': {'status': 'NOT_FOUND'}}), + json.dumps({'error': {'status': 'INTERNAL'}})], + [404, 500], + recorder + ) + functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + + queue = functions.task_queue('test-function-name', 'my-instance') + with warnings.catch_warnings(record=True) as record: + warnings.simplefilter("always") + with pytest.raises(firebase_admin.exceptions.InternalError): + queue.enqueue(_DEFAULT_DATA) + assert queue._scope.type == 'extension_or_kit' + user_warnings = [w for w in record if issubclass(w.category, UserWarning)] + assert len(user_warnings) == 0 + + def test_delete_fallback_failure_reverts_scope(self): + functions_service = functions._get_functions_service(firebase_admin.get_app()) + recorder = [] + adapter = testutils.MockMultiRequestAdapter( + [json.dumps({'error': {'status': 'NOT_FOUND'}}), + json.dumps({'error': {'status': 'INTERNAL'}})], + [404, 500], + recorder + ) + functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + + queue = functions.task_queue('test-function-name', 'my-instance') + with warnings.catch_warnings(record=True) as record: + warnings.simplefilter("always") + with pytest.raises(firebase_admin.exceptions.InternalError): + queue.delete('task-123') + assert queue._scope.type == 'extension_or_kit' + user_warnings = [w for w in record if issubclass(w.category, UserWarning)] + assert len(user_warnings) == 0 + class TestTaskQueueOptions: _DEFAULT_TASK_OPTS = {'schedule_delay_seconds': None, 'schedule_time': None, \ @@ -431,3 +634,36 @@ def test_invalid_uri_error(self, uri): assert len(recorder) == 0 assert str(excinfo.value) == \ 'uri must be a valid RFC3986 URI string using the https or http schema.' + + +class TestFunctionScope: + def test_function_scope_current(self): + scope = functions.FunctionScope.current() + assert scope.type == 'current' + assert scope.instance_id is None + + def test_function_scope_global(self): + scope = functions.FunctionScope.global_scope() + assert scope.type == 'global' + assert scope.instance_id is None + + def test_function_scope_extension(self): + scope = functions.FunctionScope.extension('my-extension') + assert scope.type == 'extension' + assert scope.instance_id == 'my-extension' + + def test_function_scope_constructor_kit(self): + scope = functions.FunctionScope('kit', 'my-kit') + assert scope.type == 'kit' + assert scope.instance_id == 'my-kit' + + @pytest.mark.parametrize('invalid_id', [None, '', 123, []]) + def test_function_scope_invalid(self, invalid_id): + with pytest.raises(ValueError): + functions.FunctionScope.extension(invalid_id) + with pytest.raises(ValueError): + functions.FunctionScope('kit', invalid_id) + + def test_function_scope_invalid_type(self): + with pytest.raises(ValueError): + functions.FunctionScope('invalid-type') From 46ecd7dbeffd1d270dafb28b57d524c5233994a9 Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Thu, 9 Jul 2026 18:01:44 -0700 Subject: [PATCH 2/6] feat(functions): Refactor fallback retry to use private helper methods Factored out payload and HTTP request preparation in TaskQueue.enqueue and TaskQueue.delete into private helper methods _enqueue_with_scope and _delete_with_scope. This avoids recursion, class state pollution, and rollback logic, keeping self._scope unmodified until the request successfully completes. --- firebase_admin/functions.py | 97 +++++++++++++++++++++---------------- 1 file changed, 56 insertions(+), 41 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index fadc0ea2..6088853f 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -309,34 +309,22 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str: Returns: str: The ID of the task relative to this queue. """ - resolved_resource, ext_or_kit_id = self._resolve_resource(self._scope) - task = self._validate_task_options(task_data, resolved_resource, opts) - service_url = self._get_emulator_url(resolved_resource) or self._get_url( - resolved_resource, _CLOUD_TASKS_API_URL_FORMAT) - task_payload = self._update_task_payload(task, resolved_resource, ext_or_kit_id) try: - resp = self._http_client.body( - 'post', - url=service_url, - headers=_FUNCTIONS_HEADERS, - json={'task': task_payload.to_api_dict()} - ) - return self._parse_enqueue_response(resp, resolved_resource) + return self._enqueue_with_scope(task_data, self._scope, opts) except requests.exceptions.RequestException as error: is_404 = error.response is not None and error.response.status_code == 404 if self._scope.type != 'extension_or_kit' or not is_404: raise _FunctionsService.handle_functions_error(error) - # Upgrade scope to kit and retry recursively, rolling back if it throws - old_scope = self._scope - self._scope = FunctionScope('kit', self._scope.instance_id) + # Retry with kit scope + kit_scope = FunctionScope('kit', self._scope.instance_id) try: - resp = self.enqueue(task_data, opts) - self._log_fallback_warning(self._function_name, self._scope.instance_id) - return resp - except Exception: - self._scope = old_scope - raise + resp = self._enqueue_with_scope(task_data, kit_scope, opts) + except requests.exceptions.RequestException as retry_error: + raise _FunctionsService.handle_functions_error(retry_error) + self._scope = kit_scope + self._log_fallback_warning(self._function_name, self._scope.instance_id) + return resp def delete(self, task_id: str) -> None: """Deletes an enqueued task if it has not yet started. @@ -352,34 +340,61 @@ def delete(self, task_id: str) -> None: ValueError: If the input arguments are invalid. """ _Validators.check_non_empty_string('task_id', task_id) - - resolved_resource, _ = self._resolve_resource(self._scope) - emulator_url = self._get_emulator_url(resolved_resource) - if emulator_url: - service_url = emulator_url + f'/{task_id}' - else: - service_url = self._get_url( - resolved_resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}') try: - self._http_client.body( - 'delete', - url=service_url, - headers=_FUNCTIONS_HEADERS, - ) + self._delete_with_scope(task_id, self._scope) except requests.exceptions.RequestException as error: is_404 = error.response is not None and error.response.status_code == 404 if not is_404: raise _FunctionsService.handle_functions_error(error) if self._scope.type == 'extension_or_kit': - old_scope = self._scope - self._scope = FunctionScope('kit', self._scope.instance_id) + kit_scope = FunctionScope('kit', self._scope.instance_id) try: - self.delete(task_id) - self._log_fallback_warning(self._function_name, self._scope.instance_id) - except Exception: - self._scope = old_scope - raise + self._delete_with_scope(task_id, kit_scope) + except requests.exceptions.RequestException as retry_error: + is_retry_404 = ( + retry_error.response is not None + and retry_error.response.status_code == 404 + ) + if not is_retry_404: + raise _FunctionsService.handle_functions_error(retry_error) + self._scope = kit_scope + self._log_fallback_warning(self._function_name, self._scope.instance_id) + + def _enqueue_with_scope( + self, + task_data: Any, + scope: FunctionScope, + opts: Optional[TaskOptions] = None + ) -> str: + """Enqueues a task using a specific FunctionScope.""" + resolved_resource, ext_or_kit_id = self._resolve_resource(scope) + task = self._validate_task_options(task_data, resolved_resource, opts) + service_url = self._get_emulator_url(resolved_resource) or self._get_url( + resolved_resource, _CLOUD_TASKS_API_URL_FORMAT) + task_payload = self._update_task_payload(task, resolved_resource, ext_or_kit_id) + resp = self._http_client.body( + 'post', + url=service_url, + headers=_FUNCTIONS_HEADERS, + json={'task': task_payload.to_api_dict()} + ) + return self._parse_enqueue_response(resp, resolved_resource) + + def _delete_with_scope(self, task_id: str, scope: FunctionScope) -> None: + """Deletes a task using a specific FunctionScope.""" + resolved_resource, _ = self._resolve_resource(scope) + emulator_url = self._get_emulator_url(resolved_resource) + if emulator_url: + service_url = emulator_url + f'/{task_id}' + else: + service_url = self._get_url( + resolved_resource, _CLOUD_TASKS_API_URL_FORMAT + f'/{task_id}') + self._http_client.body( + 'delete', + url=service_url, + headers=_FUNCTIONS_HEADERS, + ) def _parse_resource_name(self, resource_name: str, resource_id_key: str) -> Resource: From 0d0b25974d31de8f1d9772cc0a9c7da7595c04da Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Thu, 9 Jul 2026 18:04:43 -0700 Subject: [PATCH 3/6] fix(functions): Propagate 404 errors on delete and fallback retry Updated delete to raise NotFoundError on 404 response rather than silently swallowing it, preserving original SDK behavior. Added and updated tests to assert 404 propagation and rollback. --- firebase_admin/functions.py | 16 ++++++---------- tests/test_functions.py | 25 +++++++++++++++++++++++-- 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index 6088853f..be85fc23 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -344,22 +344,18 @@ def delete(self, task_id: str) -> None: self._delete_with_scope(task_id, self._scope) except requests.exceptions.RequestException as error: is_404 = error.response is not None and error.response.status_code == 404 - if not is_404: - raise _FunctionsService.handle_functions_error(error) - - if self._scope.type == 'extension_or_kit': + if self._scope.type == 'extension_or_kit' and is_404: + # Retry with kit scope kit_scope = FunctionScope('kit', self._scope.instance_id) try: self._delete_with_scope(task_id, kit_scope) except requests.exceptions.RequestException as retry_error: - is_retry_404 = ( - retry_error.response is not None - and retry_error.response.status_code == 404 - ) - if not is_retry_404: - raise _FunctionsService.handle_functions_error(retry_error) + raise _FunctionsService.handle_functions_error(retry_error) self._scope = kit_scope self._log_fallback_warning(self._function_name, self._scope.instance_id) + return + + raise _FunctionsService.handle_functions_error(error) def _enqueue_with_scope( self, diff --git a/tests/test_functions.py b/tests/test_functions.py index 1682ca5a..3be2dfc8 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -417,12 +417,13 @@ def test_delete_fallback_flow(self): assert 'ext-my-instance-test-function-name/tasks/task-123' in recorder[0].url assert 'kit-my-instance-test-function-name/tasks/task-123' in recorder[1].url - def test_delete_ignoring_404(self): + def test_delete_propagates_404(self): _, recorder = self._instrument_functions_service( status=404, payload='{"error": {"status": "NOT_FOUND"}}') queue = functions.task_queue( 'test-function-name', scope=functions.FunctionScope.global_scope()) - queue.delete('task-123') + with pytest.raises(firebase_admin.exceptions.NotFoundError): + queue.delete('task-123') assert len(recorder) == 1 def test_enqueue_fallback_failure_reverts_scope(self): @@ -465,6 +466,26 @@ def test_delete_fallback_failure_reverts_scope(self): user_warnings = [w for w in record if issubclass(w.category, UserWarning)] assert len(user_warnings) == 0 + def test_delete_fallback_failure_reverts_scope_on_404(self): + functions_service = functions._get_functions_service(firebase_admin.get_app()) + recorder = [] + adapter = testutils.MockMultiRequestAdapter( + [json.dumps({'error': {'status': 'NOT_FOUND'}}), + json.dumps({'error': {'status': 'NOT_FOUND'}})], + [404, 404], + recorder + ) + functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + + queue = functions.task_queue('test-function-name', 'my-instance') + with warnings.catch_warnings(record=True) as record: + warnings.simplefilter("always") + with pytest.raises(firebase_admin.exceptions.NotFoundError): + queue.delete('task-123') + assert queue._scope.type == 'extension_or_kit' + user_warnings = [w for w in record if issubclass(w.category, UserWarning)] + assert len(user_warnings) == 0 + class TestTaskQueueOptions: _DEFAULT_TASK_OPTS = {'schedule_delay_seconds': None, 'schedule_time': None, \ From 647756e5a5a8f1c2d9377969a6acb1c9457528ef Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Thu, 9 Jul 2026 18:17:57 -0700 Subject: [PATCH 4/6] fix(functions): Resolve Tuple type hint issue, restore error handler, and ensure complete thread safety without locks Imported Tuple from typing and updated _resolve_resource signature to support Python versions older than 3.9. Restored the accidentally removed handle_functions_error method body. Removed self._scope mutation entirely to ensure complete thread safety without locks, using a thread-safe global set _WARNED_INSTANCES to ensure each warning is only logged once. --- firebase_admin/functions.py | 15 ++++++++++----- tests/test_functions.py | 9 +++++++-- 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index be85fc23..a06b9b7d 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -22,7 +22,7 @@ import warnings import json from base64 import b64encode -from typing import Any, Optional, Dict, Union +from typing import Any, Optional, Dict, Union, Tuple from dataclasses import dataclass from google.auth.compute_engine import Credentials as ComputeEngineCredentials @@ -196,9 +196,12 @@ def task_queue( @classmethod def handle_functions_error(cls, error: Any): """Handles errors received from the Cloud Functions API.""" - return _utils.handle_platform_error_from_requests(error) + +_WARNED_INSTANCES = set() + + class TaskQueue: """TaskQueue class that implements Firebase Cloud Tasks Queues functionality.""" def __init__( @@ -239,7 +242,7 @@ def __init__( self._resource.location_id = self._resource.location_id or _DEFAULT_LOCATION _Validators.check_non_empty_string('resource.resource_id', self._resource.resource_id) - def _resolve_resource(self, scope: FunctionScope) -> tuple[Resource, Optional[str]]: + def _resolve_resource(self, scope: FunctionScope) -> Tuple[Resource, Optional[str]]: """Resolves the Resource object and extension/kit ID based on the scope.""" resource_id = self._resource.resource_id extension_or_kit_id = None @@ -267,6 +270,10 @@ def _resolve_resource(self, scope: FunctionScope) -> tuple[Resource, Optional[st def _log_fallback_warning(self, function_name: str, instance: str) -> None: """Logs a warning when falling back from extension to kit.""" + key = (function_name, instance) + if key in _WARNED_INSTANCES: + return + _WARNED_INSTANCES.add(key) warnings.warn( f"Targeting kit {instance} with the legacy extensions API, " "which has performance implications. Please change the call " @@ -322,7 +329,6 @@ def enqueue(self, task_data: Any, opts: Optional[TaskOptions] = None) -> str: resp = self._enqueue_with_scope(task_data, kit_scope, opts) except requests.exceptions.RequestException as retry_error: raise _FunctionsService.handle_functions_error(retry_error) - self._scope = kit_scope self._log_fallback_warning(self._function_name, self._scope.instance_id) return resp @@ -351,7 +357,6 @@ def delete(self, task_id: str) -> None: self._delete_with_scope(task_id, kit_scope) except requests.exceptions.RequestException as retry_error: raise _FunctionsService.handle_functions_error(retry_error) - self._scope = kit_scope self._log_fallback_warning(self._function_name, self._scope.instance_id) return diff --git a/tests/test_functions.py b/tests/test_functions.py index 3be2dfc8..124b9f98 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -385,12 +385,13 @@ def test_enqueue_fallback_flow(self): recorder ) functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + functions._WARNED_INSTANCES.clear() queue = functions.task_queue('test-function-name', 'my-instance') with pytest.warns(UserWarning) as record: task_id = queue.enqueue(_DEFAULT_DATA) assert task_id == 'task-123' - assert queue._scope.type == 'kit' + assert queue._scope.type == 'extension_or_kit' assert len(record) == 1 assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) assert len(recorder) == 2 @@ -406,11 +407,12 @@ def test_delete_fallback_flow(self): recorder ) functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + functions._WARNED_INSTANCES.clear() queue = functions.task_queue('test-function-name', 'my-instance') with pytest.warns(UserWarning) as record: queue.delete('task-123') - assert queue._scope.type == 'kit' + assert queue._scope.type == 'extension_or_kit' assert len(record) == 1 assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) assert len(recorder) == 2 @@ -436,6 +438,7 @@ def test_enqueue_fallback_failure_reverts_scope(self): recorder ) functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + functions._WARNED_INSTANCES.clear() queue = functions.task_queue('test-function-name', 'my-instance') with warnings.catch_warnings(record=True) as record: @@ -456,6 +459,7 @@ def test_delete_fallback_failure_reverts_scope(self): recorder ) functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + functions._WARNED_INSTANCES.clear() queue = functions.task_queue('test-function-name', 'my-instance') with warnings.catch_warnings(record=True) as record: @@ -476,6 +480,7 @@ def test_delete_fallback_failure_reverts_scope_on_404(self): recorder ) functions_service._http_client.session.mount(_CLOUD_TASKS_URL, adapter) + functions._WARNED_INSTANCES.clear() queue = functions.task_queue('test-function-name', 'my-instance') with warnings.catch_warnings(record=True) as record: From cf3590f376fc70387bb8a12dfd2f9bcfac957ccb Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Fri, 10 Jul 2026 08:02:51 -0700 Subject: [PATCH 5/6] feat(functions): Rename KIT_INSTANCE_ID env var to FIREBASE_KIT_INSTANCE_ID Renamed the KIT_INSTANCE_ID environment variable to FIREBASE_KIT_INSTANCE_ID in both functions.py and tests. --- firebase_admin/functions.py | 2 +- tests/test_functions.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index a06b9b7d..8f1b8f00 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -248,7 +248,7 @@ def _resolve_resource(self, scope: FunctionScope) -> Tuple[Resource, Optional[st extension_or_kit_id = None if scope.type == 'current': - kit_instance_id = os.environ.get('KIT_INSTANCE_ID') + kit_instance_id = os.environ.get('FIREBASE_KIT_INSTANCE_ID') if kit_instance_id: resource_id = f'kit-{kit_instance_id}-{resource_id}' extension_or_kit_id = kit_instance_id diff --git a/tests/test_functions.py b/tests/test_functions.py index 124b9f98..dafa73f4 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -303,7 +303,7 @@ def test_enqueue_current_scope_env_vars_ext(self, monkeypatch): assert recorder[0].url == expected_url def test_enqueue_current_scope_env_vars_kit(self, monkeypatch): - monkeypatch.setenv('KIT_INSTANCE_ID', 'my-kit') + monkeypatch.setenv('FIREBASE_KIT_INSTANCE_ID', 'my-kit') expected_queue = 'kit-my-kit-test-function-name' response_payload = json.dumps({ 'name': ( From 9f3d0822ff2c3bf2537d284bbb908e19cf3e69af Mon Sep 17 00:00:00 2001 From: Thomas Bouldin Date: Fri, 10 Jul 2026 09:14:45 -0700 Subject: [PATCH 6/6] fix(functions): Support EXT_INSTANCE_ID in current scope and improve fallback warning suggestion Updated _resolve_resource to support resolving current scope using EXT_INSTANCE_ID (checking it first before falling back to FIREBASE_KIT_INSTANCE_ID) for completeness and consistency. Updated _log_fallback_warning message to suggest the more idiomatic functions.FunctionScope. --- firebase_admin/functions.py | 8 ++++++-- tests/test_functions.py | 35 +++++++++++++++++++++++++++++++---- 2 files changed, 37 insertions(+), 6 deletions(-) diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py index 8f1b8f00..a28ac07b 100644 --- a/firebase_admin/functions.py +++ b/firebase_admin/functions.py @@ -248,8 +248,12 @@ def _resolve_resource(self, scope: FunctionScope) -> Tuple[Resource, Optional[st extension_or_kit_id = None if scope.type == 'current': + ext_instance_id = os.environ.get('EXT_INSTANCE_ID') kit_instance_id = os.environ.get('FIREBASE_KIT_INSTANCE_ID') - if kit_instance_id: + if ext_instance_id: + resource_id = f'ext-{ext_instance_id}-{resource_id}' + extension_or_kit_id = ext_instance_id + elif kit_instance_id: resource_id = f'kit-{kit_instance_id}-{resource_id}' extension_or_kit_id = kit_instance_id elif scope.type == 'global': @@ -278,7 +282,7 @@ def _log_fallback_warning(self, function_name: str, instance: str) -> None: f"Targeting kit {instance} with the legacy extensions API, " "which has performance implications. Please change the call " f"task_queue('{function_name}', '{instance}') to " - f"task_queue('{function_name}', FunctionScope('kit', '{instance}'))", + f"task_queue('{function_name}', functions.FunctionScope('kit', '{instance}'))", UserWarning, stacklevel=3 ) diff --git a/tests/test_functions.py b/tests/test_functions.py index dafa73f4..4a8aadfb 100644 --- a/tests/test_functions.py +++ b/tests/test_functions.py @@ -290,15 +290,21 @@ def test_enqueue_current_scope_default(self): assert recorder[0].url == expected_url def test_enqueue_current_scope_env_vars_ext(self, monkeypatch): - # Python SDK does not handle EXT_INSTANCE_ID since no extensions use Python monkeypatch.setenv('EXT_INSTANCE_ID', 'my-ext') - _, recorder = self._instrument_functions_service() + expected_queue = 'ext-my-ext-test-function-name' + response_payload = json.dumps({ + 'name': ( + f'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks/test-task-id' + ) + }) + _, recorder = self._instrument_functions_service(payload=response_payload) queue = functions.task_queue('test-function-name') queue.enqueue(_DEFAULT_DATA) assert len(recorder) == 1 expected_url = ( _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' - 'queues/test-function-name/tasks' + f'queues/{expected_queue}/tasks' ) assert recorder[0].url == expected_url @@ -321,6 +327,26 @@ def test_enqueue_current_scope_env_vars_kit(self, monkeypatch): ) assert recorder[0].url == expected_url + def test_enqueue_current_scope_env_vars_precedence(self, monkeypatch): + monkeypatch.setenv('EXT_INSTANCE_ID', 'my-ext') + monkeypatch.setenv('FIREBASE_KIT_INSTANCE_ID', 'my-kit') + expected_queue = 'ext-my-ext-test-function-name' + response_payload = json.dumps({ + 'name': ( + f'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks/test-task-id' + ) + }) + _, recorder = self._instrument_functions_service(payload=response_payload) + queue = functions.task_queue('test-function-name') + queue.enqueue(_DEFAULT_DATA) + assert len(recorder) == 1 + expected_url = ( + _CLOUD_TASKS_URL + 'projects/test-project/locations/us-central1/' + f'queues/{expected_queue}/tasks' + ) + assert recorder[0].url == expected_url + def test_enqueue_global_scope(self): _, recorder = self._instrument_functions_service() queue = functions.task_queue( @@ -392,8 +418,8 @@ def test_enqueue_fallback_flow(self): task_id = queue.enqueue(_DEFAULT_DATA) assert task_id == 'task-123' assert queue._scope.type == 'extension_or_kit' - assert len(record) == 1 assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) + assert "functions.FunctionScope('kit', 'my-instance')" in str(record[0].message) assert len(recorder) == 2 assert 'ext-my-instance-test-function-name' in recorder[0].url assert 'kit-my-instance-test-function-name' in recorder[1].url @@ -415,6 +441,7 @@ def test_delete_fallback_flow(self): assert queue._scope.type == 'extension_or_kit' assert len(record) == 1 assert "Targeting kit my-instance with the legacy extensions API" in str(record[0].message) + assert "functions.FunctionScope('kit', 'my-instance')" in str(record[0].message) assert len(recorder) == 2 assert 'ext-my-instance-test-function-name/tasks/task-123' in recorder[0].url assert 'kit-my-instance-test-function-name/tasks/task-123' in recorder[1].url