feat(python): switch to uv and ruff for build tooling

Switched the Python samples' build tooling to uv and ruff.

- Updated `pyproject.toml` to use ruff for linting and formatting.
- Updated `requirements.txt` to include uv and ruff.
- Deleted the custom `pyfmt.py` script.
- Updated the GitHub Actions workflow to use uv and ruff.
- Fixed all linting errors in the Python samples.
- Maintained pip compatibility for deployment.

Author: google-labs-jules[bot]

.github/workflows/test_python.yml

@@ -42,9 +42,18 @@ jobs:
         uses: actions/setup-python@v4
         with:
             python-version: ${{ matrix.python-version }}
-      - name: Install dependencies
+      - uses: astral-sh/setup-uv@v1
+      - name: Install dev dependencies
         working-directory: ./Python
-        run: pip install -r requirements.txt
+        run: uv pip install -r requirements.txt
       - name: Lint
         working-directory: ./Python
-        run: python pyfmt.py --check_only --exclude "**/venv/**/*.py" **/*.py
+        run: |
+          ruff format --check .
+          ruff check .
+      - name: Check pip compatibility by installing sample dependencies
+        working-directory: ./Python
+        run: |
+          find . -mindepth 2 -name "requirements.txt" -print0 | while IFS= read -r -d $'\0' file; do
+            python -m pip install -r "$file"
+          done

Python/alerts-to-discord/functions/main.py

@@ -14,18 +14,19 @@
 
 import pprint
 
+# [END v2import]
+import requests
+
 # [START v2import]
 from firebase_functions import params
 from firebase_functions.alerts import app_distribution_fn, crashlytics_fn, performance_fn
-# [END v2import]
-
-import requests
 
 DISCORD_WEBHOOK_URL = params.SecretParam("DISCORD_WEBHOOK_URL")
 
 
-def post_message_to_discord(bot_name: str, message_body: str,
-                            webhook_url: str) -> requests.Response:
+def post_message_to_discord(
+    bot_name: str, message_body: str, webhook_url: str
+) -> requests.Response:
     """Posts a message to Discord with Discord's Webhook API.
 
     Params:
@@ -36,24 +37,26 @@ def post_message_to_discord(bot_name: str, message_body: str,
         raise EnvironmentError(
             "No webhook URL found. Set the Discord Webhook URL before deploying. "
             "Learn more about Discord webhooks here: "
-            "https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks")
+            "https://support.discord.com/hc/en-us/articles/228383668-Intro-to-Webhooks"
+        )
 
     return requests.post(
         url=webhook_url,
         json={
             # Here's what the Discord API supports in the payload:
             # https://discord.com/developers/docs/resources/webhook#execute-webhook-jsonform-params
             "username": bot_name,
-            "content": message_body
-        })
+            "content": message_body,
+        },
+    )
 
 
 # [START v2Alerts]
 # [START v2CrashlyticsAlertTrigger]
 @crashlytics_fn.on_new_fatal_issue_published(secrets=["DISCORD_WEBHOOK_URL"])
 def post_fatal_issue_to_discord(event: crashlytics_fn.CrashlyticsNewFatalIssueEvent) -> None:
     """Publishes a message to Discord whenever a new Crashlytics fatal issue occurs."""
-# [END v2CrashlyticsAlertTrigger]
+    # [END v2CrashlyticsAlertTrigger]
     # [START v2CrashlyticsEventPayload]
     # Construct a helpful message to send to Discord.
     app_id = event.app_id
@@ -86,7 +89,7 @@ def post_fatal_issue_to_discord(event: crashlytics_fn.CrashlyticsNewFatalIssueEv
 @app_distribution_fn.on_new_tester_ios_device_published(secrets=["DISCORD_WEBHOOK_URL"])
 def post_new_udid_to_discord(event: app_distribution_fn.NewTesterDeviceEvent) -> None:
     """Publishes a message to Discord whenever someone registers a new iOS test device."""
-# [END v2AppDistributionAlertTrigger]
+    # [END v2AppDistributionAlertTrigger]
     # [START v2AppDistributionEventPayload]
     # Construct a helpful message to send to Discord.
     app_id = event.app_id
@@ -110,14 +113,15 @@ def post_new_udid_to_discord(event: app_distribution_fn.NewTesterDeviceEvent) ->
     except (EnvironmentError, requests.HTTPError) as error:
         print(
             f"Unable to post iOS device registration alert for {app_dist.tester_email} to Discord.",
-            error)
+            error,
+        )
 
 
 # [START v2PerformanceAlertTrigger]
 @performance_fn.on_threshold_alert_published(secrets=["DISCORD_WEBHOOK_URL"])
 def post_performance_alert_to_discord(event: performance_fn.PerformanceThresholdAlertEvent) -> None:
     """Publishes a message to Discord whenever a performance threshold alert is fired."""
-# [END v2PerformanceAlertTrigger]
+    # [END v2PerformanceAlertTrigger]
     # [START v2PerformanceEventPayload]
     # Construct a helpful message to send to Discord.
     app_id = event.app_id
@@ -139,8 +143,9 @@ def post_performance_alert_to_discord(event: performance_fn.PerformanceThreshold
 
     try:
         # [START v2SendPerformanceAlertToDiscord]
-        response = post_message_to_discord("App Performance Bot", message,
-                                           DISCORD_WEBHOOK_URL.value)
+        response = post_message_to_discord(
+            "App Performance Bot", message, DISCORD_WEBHOOK_URL.value
+        )
         if response.ok:
             print(f"Posted Firebase Performance alert {perf.event_name} to Discord.")
             pprint.pp(event.data.payload)
@@ -149,4 +154,6 @@ def post_performance_alert_to_discord(event: performance_fn.PerformanceThreshold
         # [END v2SendPerformanceAlertToDiscord]
     except (EnvironmentError, requests.HTTPError) as error:
         print(f"Unable to post Firebase Performance alert {perf.event_name} to Discord.", error)
+
+
 # [END v2Alerts]

Python/delete-unused-accounts-cron/functions/main.py

@@ -15,14 +15,14 @@
 # [START all]
 from datetime import datetime, timedelta
 
-# [START import]
-# The Cloud Functions for Firebase SDK to set up triggers and logging.
-from firebase_functions import scheduler_fn
-
 # The Firebase Admin SDK to delete users.
 import firebase_admin
 from firebase_admin import auth
 
+# [START import]
+# The Cloud Functions for Firebase SDK to set up triggers and logging.
+from firebase_functions import scheduler_fn
+
 firebase_admin.initialize_app()
 # [END import]
 
@@ -40,6 +40,8 @@ def accountcleanup(event: scheduler_fn.ScheduledEvent) -> None:
         ]
         auth.delete_users(inactive_uids)
         user_page = user_page.get_next_page()
+
+
 # [END accountcleanup]
 
 
@@ -55,4 +57,6 @@ def is_inactive(user: auth.UserRecord, inactive_limit: timedelta) -> bool:
     last_seen = datetime.fromtimestamp(last_seen_timestamp)
     inactive_time = datetime.now() - last_seen
     return inactive_time >= inactive_limit
+
+
 # [END all]

Python/fcm-notifications/functions/main.py

@@ -1,5 +1,5 @@
 import firebase_admin
-from firebase_admin import auth, db, messaging, exceptions
+from firebase_admin import auth, db, exceptions, messaging
 from firebase_functions import db_fn
 
 firebase_admin.initialize_app()
@@ -25,7 +25,7 @@ def send_follower_notification(event: db_fn.Event[db_fn.Change]) -> None:
     print(f"User {follower_uid} is now following user {followed_uid}")
     tokens_ref = db.reference(f"users/{followed_uid}/notificationTokens")
     notification_tokens = tokens_ref.get()
-    if (not isinstance(notification_tokens, dict) or len(notification_tokens) < 1):
+    if not isinstance(notification_tokens, dict) or len(notification_tokens) < 1:
         print("There are no tokens to send notifications to.")
         return
     print(f"There are {len(notification_tokens)} tokens to send notifications to.")
@@ -52,6 +52,8 @@ def send_follower_notification(event: db_fn.Event[db_fn.Change]) -> None:
         if not isinstance(exception, exceptions.FirebaseError):
             continue
         message = exception.http_response.json()["error"]["message"]
-        if (isinstance(exception, messaging.UnregisteredError) or
-                message == "The registration token is not a valid FCM registration token"):
+        if (
+            isinstance(exception, messaging.UnregisteredError)
+            or message == "The registration token is not a valid FCM registration token"
+        ):
             tokens_ref.child(msgs[i].token).delete()

Python/http-flask/functions/main.py

@@ -13,9 +13,9 @@
 # limitations under the License.
 
 # [START httpflaskexample]
-from firebase_admin import initialize_app, db
-from firebase_functions import https_fn
 import flask
+from firebase_admin import db, initialize_app
+from firebase_functions import https_fn
 
 initialize_app()
 app = flask.Flask(__name__)
@@ -46,4 +46,6 @@ def add_widget():
 def httpsflaskexample(req: https_fn.Request) -> https_fn.Response:
     with app.request_context(req.environ):
         return app.full_dispatch_request()
+
+
 # [END httpflaskexample]

Python/post-signup-event/functions/main.py

@@ -12,26 +12,26 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from datetime import datetime, timedelta
 import json
-
-from firebase_admin import auth, firestore, initialize_app
-from firebase_functions import https_fn, identity_fn, tasks_fn, options, params
+from datetime import datetime, timedelta
 
 import google.auth
 import google.auth.transport.requests
 import google.cloud.firestore
 import google.cloud.tasks_v2
 import google.oauth2.credentials
 import googleapiclient.discovery
+from firebase_admin import auth, firestore, initialize_app
+from firebase_functions import https_fn, identity_fn, options, params, tasks_fn
 
 initialize_app()
 
 
 # [START savegoogletoken]
 @identity_fn.before_user_created()
 def savegoogletoken(
-        event: identity_fn.AuthBlockingEvent) -> identity_fn.BeforeCreateResponse | None:
+    event: identity_fn.AuthBlockingEvent,
+) -> identity_fn.BeforeCreateResponse | None:
     """During sign-up, save the Google OAuth2 access token and queue up a task
     to schedule an onboarding session on the user's Google Calendar.
 
@@ -48,25 +48,22 @@ def savegoogletoken(
         doc_ref.set({"calendar_access_token": event.credential.access_token}, merge=True)
 
         tasks_client = google.cloud.tasks_v2.CloudTasksClient()
-        task_queue = tasks_client.queue_path(params.PROJECT_ID.value,
-                                             options.SupportedRegion.US_CENTRAL1,
-                                             "scheduleonboarding")
+        task_queue = tasks_client.queue_path(
+            params.PROJECT_ID.value, options.SupportedRegion.US_CENTRAL1, "scheduleonboarding"
+        )
         target_uri = get_function_url("scheduleonboarding")
-        calendar_task = google.cloud.tasks_v2.Task(http_request={
-            "http_method": google.cloud.tasks_v2.HttpMethod.POST,
-            "url": target_uri,
-            "headers": {
-                "Content-type": "application/json"
+        calendar_task = google.cloud.tasks_v2.Task(
+            http_request={
+                "http_method": google.cloud.tasks_v2.HttpMethod.POST,
+                "url": target_uri,
+                "headers": {"Content-type": "application/json"},
+                "body": json.dumps({"data": {"uid": event.data.uid}}).encode(),
             },
-            "body": json.dumps({
-                "data": {
-                    "uid": event.data.uid
-                }
-            }).encode()
-        },
-                                                   schedule_time=datetime.now() +
-                                                   timedelta(minutes=1))
+            schedule_time=datetime.now() + timedelta(minutes=1),
+        )
         tasks_client.create_task(parent=task_queue, task=calendar_task)
+
+
 # [END savegoogletoken]
 
 
@@ -79,50 +76,54 @@ def scheduleonboarding(request: tasks_fn.CallableRequest) -> https_fn.Response:
     """
 
     if "uid" not in request.data:
-        return https_fn.Response(status=https_fn.FunctionsErrorCode.INVALID_ARGUMENT,
-                                 response="No user specified.")
+        return https_fn.Response(
+            status=https_fn.FunctionsErrorCode.INVALID_ARGUMENT, response="No user specified."
+        )
     uid = request.data["uid"]
 
     user_record: auth.UserRecord = auth.get_user(uid)
     if user_record.email is None:
-        return https_fn.Response(status=https_fn.FunctionsErrorCode.INVALID_ARGUMENT,
-                                 response="No email address on record.")
+        return https_fn.Response(
+            status=https_fn.FunctionsErrorCode.INVALID_ARGUMENT,
+            response="No email address on record.",
+        )
 
     firestore_client: google.cloud.firestore.Client = firestore.client()
     user_info = firestore_client.collection("user_info").document(uid).get().to_dict()
     if not isinstance(user_info, dict) or "calendar_access_token" not in user_info:
-        return https_fn.Response(status=https_fn.FunctionsErrorCode.PERMISSION_DENIED,
-                                 response="No Google OAuth token found.")
+        return https_fn.Response(
+            status=https_fn.FunctionsErrorCode.PERMISSION_DENIED,
+            response="No Google OAuth token found.",
+        )
     calendar_access_token = user_info["calendar_access_token"]
     firestore_client.collection("user_info").document(uid).update(
-        {"calendar_access_token": google.cloud.firestore.DELETE_FIELD})
+        {"calendar_access_token": google.cloud.firestore.DELETE_FIELD}
+    )
 
     google_credentials = google.oauth2.credentials.Credentials(token=calendar_access_token)
 
-    calendar_client = googleapiclient.discovery.build("calendar",
-                                                      "v3",
-                                                      credentials=google_credentials)
+    calendar_client = googleapiclient.discovery.build(
+        "calendar", "v3", credentials=google_credentials
+    )
     calendar_event = {
         "summary": "Onboarding with ExampleCo",
         "location": "Video call",
         "description": "Walk through onboarding tasks with an ExampleCo engineer.",
         "start": {
             "dateTime": (datetime.now() + timedelta(days=3)).isoformat(),
-            "timeZone": "America/Los_Angeles"
+            "timeZone": "America/Los_Angeles",
         },
         "end": {
             "dateTime": (datetime.now() + timedelta(days=3, hours=1)).isoformat(),
-            "timeZone": "America/Los_Angeles"
+            "timeZone": "America/Los_Angeles",
         },
-        "attendees": [{
-            "email": user_record.email
-        }, {
-            "email": "[email protected]"
-        }]
+        "attendees": [{"email": user_record.email}, {"email": "[email protected]"}],
     }
     calendar_client.events().insert(calendarId="primary", body=calendar_event).execute()
 
     return https_fn.Response("Success")
+
+
 # [END scheduleonboarding]
 
 
@@ -137,10 +138,13 @@ def get_function_url(name: str, location: str = options.SupportedRegion.US_CENTR
         The URL of the function
     """
     credentials, project_id = google.auth.default(
-        scopes=["https://www.googleapis.com/auth/cloud-platform"])
+        scopes=["https://www.googleapis.com/auth/cloud-platform"]
+    )
     authed_session = google.auth.transport.requests.AuthorizedSession(credentials)
-    url = ("https://cloudfunctions.googleapis.com/v2beta/" +
-           f"projects/{project_id}/locations/{location}/functions/{name}")
+    url = (
+        "https://cloudfunctions.googleapis.com/v2beta/"
+        + f"projects/{project_id}/locations/{location}/functions/{name}"
+    )
     response = authed_session.get(url)
     data = response.json()
     function_url = data["serviceConfig"]["uri"]