Firebase Sdk

Firebase Admin Python SDK Integration

Integrates Firebase using firebase-admin 7.x with patterns for Firestore document CRUD and queries, Realtime Database operations, Auth (custom tokens, ID token verification, user management), Firebase Cloud Messaging (FCM) push notifications, and Firebase Storage.

TL;DR Checklist

• Use firebase_admin.initialize_app() with a service account credentials JSON to start
• Use firestore.client() for Firestore operations (preferred over Realtime DB for structured data)
• Use db.reference() for Realtime Database (preferred for real-time sync and simple data)
• Use auth.verify_id_token() to authenticate requests from client SDKs
• Use auth.create_custom_token() for generating custom auth tokens
• Use messaging.send_each() for batch FCM notifications (not send_all() which is deprecated)
• Use bucket() for Firebase Storage (Cloud Storage) file operations

When to Use

Use this skill when:

• Building server-side logic for Firebase-powered mobile or web applications
• Verifying Firebase ID tokens in your backend API to authenticate users
• Managing Firebase users (create, update, delete, list) from a server environment
• Sending push notifications via Firebase Cloud Messaging (FCM)
• Performing Firestore queries (collections, documents, composite filters, aggregations)
• Managing Realtime Database data with server-side write access
• Working with Firebase Storage buckets for file management
• Implementing custom authentication flows with Firebase Auth

When NOT to Use

• For client-side Firebase operations (use the Firebase client SDKs — JS, Android, iOS)
• When you need complex relational queries with joins (use PostgreSQL instead)
• For applications that don't need Google Cloud integration (use a simpler auth solution)
• For high-throughput analytics queries (use BigQuery)
• When you need full SQL flexibility (Firestore is a NoSQL document store)

Core Workflow

1. Initialize Firebase Admin SDK

import firebase_admin
from firebase_admin import credentials, firestore, auth, messaging, db, storage
from firebase_admin.firestore import firestore as firestore_module
from google.cloud.firestore import (
    DocumentSnapshot, CollectionReference, DocumentReference,
    Increment, ArrayUnion, ArrayRemove,
)
import os

# Option A: Service account JSON file
cred = credentials.Certificate("path/to/service-account-key.json")

# Option B: Service account from environment variable
import json
cred = credentials.Certificate(json.loads(os.environ["FIREBASE_SERVICE_ACCOUNT_JSON"]))

firebase_admin.initialize_app(cred, {
    "projectId": "my-firebase-project",
    "databaseURL": "https://my-firebase-project-default-rtdb.firebaseio.com",
    "storageBucket": "my-firebase-project.appspot.com",
})

Checkpoint: Verify initialization by calling firebase_admin.get_app() — it raises ValueError if not initialized. Use firebase_admin.initialize_app() only once per process; call firebase_admin.delete_app() in shutdown hooks.

2. Firestore CRUD Operations

def create_user_profile(db: firestore_module.Client, user_id: str, profile: dict) -> DocumentReference:
    """Create or overwrite a user profile document."""
    doc_ref = db.collection("users").document(user_id)
    doc_ref.set({
        **profile,
        "created_at": firestore.SERVER_TIMESTAMP,
        "updated_at": firestore.SERVER_TIMESTAMP,
    })
    return doc_ref

def get_user_profile(db: firestore_module.Client, user_id: str) -> dict | None:
    """Retrieve a user profile document."""
    doc: DocumentSnapshot = db.collection("users").document(user_id).get()
    return doc.to_dict() if doc.exists else None

def update_user_profile(db: firestore_module.Client, user_id: str, updates: dict) -> None:
    """Update specific fields on a user profile."""
    db.collection("users").document(user_id).update({
        **updates,
        "updated_at": firestore.SERVER_TIMESTAMP,
    })

def query_active_users(db: firestore_module.Client, min_age: int = 18, limit: int = 100) -> list[dict]:
    """Query users with composite filters."""
    docs = (
        db.collection("users")
        .where("status", "==", "active")
        .where("age", ">=", min_age)
        .order_by("age", direction=firestore.Query.ASCENDING)
        .limit(limit)
        .stream()
    )
    return [{"id": doc.id, **doc.to_dict()} for doc in docs]

Checkpoint: Firestore requires composite indexes for queries with multiple where clauses. Use SERVER_TIMESTAMP for time fields — it uses Google Cloud's clock, not the client's. Use update() for partial updates and set() with merge=True for upserts.

3. Authentication — Verify ID Tokens

def verify_firebase_token(id_token: str) -> dict:
    """Verify a Firebase ID token and return decoded claims."""
    try:
        decoded_token = auth.verify_id_token(
            id_token,
            check_revoked=True,  # Checks if the token has been revoked
        )
        return {
            "uid": decoded_token["uid"],
            "email": decoded_token.get("email"),
            "email_verified": decoded_token.get("email_verified", False),
            "phone_number": decoded_token.get("phone_number"),
            "auth_time": decoded_token.get("auth_time"),
        }
    except auth.ExpiredIdTokenError:
        raise PermissionError("Token has expired — client must refresh")
    except auth.RevokedIdTokenError:
        raise PermissionError("Token has been revoked — user may need to re-authenticate")
    except auth.InvalidIdTokenError:
        raise PermissionError("Invalid token format or signature")

Checkpoint: Always verify tokens on every API request that requires authentication. Set check_revoked=True to catch revoked sessions. Cache the public key certificates (auth.Emulator for testing) to avoid re-fetching on every call.

4. Firebase Cloud Messaging (FCM)

def send_push_notification(
    registration_tokens: list[str],
    title: str,
    body: str,
    data: dict | None = None,
) -> dict:
    """Send push notifications to device tokens."""
    message = messaging.MulticastMessage(
        tokens=registration_tokens,
        notification=messaging.Notification(
            title=title,
            body=body,
        ),
        data={k: str(v) for k, v in (data or {}).items()},
        android=messaging.AndroidConfig(
            priority="high",
            ttl=3600,
        ),
        apns=messaging.APNSConfig(
            payload=messaging.APNSPayload(
                aps=messaging.Aps(
                    sound="default",
                    badge=1,
                    content_available=True,
                ),
            ),
        ),
    )

    response = messaging.send_each_for_multicast(message)
    return {
        "success_count": response.success_count,
        "failure_count": response.failure_count,
        "invalid_tokens": [
            tokens[idx]
            for idx, resp in enumerate(response.responses)
            if isinstance(resp.exception, messaging.UnregisteredError)
        ],
    }

Checkpoint: FCM registration tokens expire. Track UnregisteredError responses and remove invalid tokens from your database. Use send_each_for_multicast() for batch sends (up to 500 tokens per call).

Implementation Patterns

Pattern 1: Custom Token Generation

def create_custom_auth_token(uid: str, claims: dict | None = None) -> str:
    """Create a custom Firebase token for your own auth system."""
    custom_token = auth.create_custom_token(
        uid,
        developer_claims=claims,
    )
    return custom_token.decode("utf-8") if isinstance(custom_token, bytes) else custom_token

Pattern 2: User Management

def list_all_users(max_results: int = 100) -> list[dict]:
    """List all Firebase Auth users with pagination."""
    users = []
    page = auth.list_users(max_results=max_results)
    while page:
        for user in page.users:
            users.append({
                "uid": user.uid,
                "email": user.email,
                "disabled": user.disabled,
                "created_at": user.user_metadata.creation_timestamp,
            })
        page = page.get_next_page()
    return users

def delete_inactive_users(inactive_uids: list[str]) -> dict:
    """Batch delete inactive users."""
    result = auth.delete_users(inactive_uids)
    return {
        "success_count": len(result.success_count),
        "failure_count": len(result.errors),
        "errors": [(err.index, err.reason) for err in result.errors],
    }

Pattern 3: Firestore Aggregation Query

def count_active_users_by_region(db: firestore_module.Client, region: str) -> int:
    """Count active users in a region using a composite query."""
    query = (
        db.collection("users")
        .where("status", "==", "active")
        .where("region", "==", region)
    )
    # Firestore count() aggregation (requires Firebase Rules index)
    from google.cloud.firestore import aggregation
    result = query.count().get()
    return result[0].value if result else 0

BAD vs GOOD: Token Verification

# ❌ BAD — No token verification, trusting client UID directly
def delete_user_bad(uid: str):
    auth.delete_user(uid)  # Anyone can call this!

# ✅ GOOD — Verify the token first, then check the UID matches
def delete_user_good(id_token: str, target_uid: str):
    claims = verify_firebase_token(id_token)
    if claims["uid"] != target_uid and "admin" not in claims.get("role", []):
        raise PermissionError("Cannot delete other users")
    auth.delete_user(target_uid)

BAD vs GOOD: Firestore Writes

# ❌ BAD — Overwriting entire document instead of updating fields
def update_email_bad(db, uid: str, new_email: str):
    doc = db.collection("users").document(uid).get()
    data = doc.to_dict()
    data["email"] = new_email
    db.collection("users").document(uid).set(data)  # Loses any new fields!

# ✅ GOOD — Partial update with update()
def update_email_good(db, uid: str, new_email: str):
    db.collection("users").document(uid).update({
        "email": new_email,
        "email_verified": False,
        "updated_at": firestore.SERVER_TIMESTAMP,
    })

Constraints

MUST DO

• Verify Firebase ID tokens on every authenticated backend request — never trust client-supplied UIDs
• Use check_revoked=True in verify_id_token() to detect session revocation
• Use firestore.SERVER_TIMESTAMP for all timestamp fields — prevents clock skew issues
• Use update() for partial Firestore document updates — not set() overwrite
• Handle UnregisteredError in FCM responses and remove invalid tokens
• Store FCM registration tokens in Firestore with a TTL/cleanup strategy
• Use composite indexes for Firestore queries with multiple where clauses

MUST NOT DO

• Never use the Admin SDK on client devices — it has full access to all Firebase services
• Do not generate custom tokens without verifying the end-user's identity first
• Avoid send_all() for FCM — it's deprecated in favor of send_each() and send_multicast()
• Never expose service account credentials in version control, client code, or logs
• Do not write Firestore documents with client-supplied timestamps — use SERVER_TIMESTAMP
• Avoid deeply nested Firestore collections (>2 levels) — they affect query performance

Output Template

When writing Firebase Admin SDK integration code, structure your output as:

1. Initialization — firebase_admin.initialize_app() with credentials and config
2. Service Client — firestore.client(), auth, messaging, db (Realtime), or bucket()
3. Authentication — Verify ID token or check custom claims before any user operation
4. Operation — Firestore CRUD, FCM send, or Auth user management
5. Error Handling — Catch firebase_admin.exceptions subclasses (UnauthenticatedError, NotFoundError, etc.)

Related Skills

Live References

• Firebase Admin Python SDK Docs — Official firebase-admin API reference
• Firestore Python Client — Firestore document operations
• Firebase Auth Admin API — User management and token verification
• Firebase Cloud Messaging — FCM notification sending
• Firebase Admin SDK Setup — Initialization and configuration
• Firebase Auth Emulator — Local testing without production Firebase
• Firebase Admin Python GitHub — Source code and release notes