name: firebase-sdk description: Integrates Firebase using firebase-admin 7.x with patterns for Firestore CRUD, Realtime Database, Auth (token verification, user management), Cloud Messaging, and hosting. license: MIT compatibility: opencode metadata: version: "1.0.0" domain: coding triggers: firebase, firebase admin, firestore, firebase auth, how do i use firebase from python, firebase realtime database, fcm, firebase cloud messaging archetypes:
- tactical
- generation anti_triggers:
- brainstorming
- vague ideation
- code golf
- over-engineering response_profile: verbosity: low directive_strength: high abstraction_level: operational role: implementation scope: implementation output-format: code content-types:
- code
- guidance
- do-dont
- examples related-skills: coding-supabase-sdk, coding-postgresql-sdk, coding-authentication-patterns
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=Trueinverify_id_token()to detect session revocation - Use
firestore.SERVER_TIMESTAMPfor all timestamp fields — prevents clock skew issues - Use
update()for partial Firestore document updates — notset()overwrite - Handle
UnregisteredErrorin 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
whereclauses
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 ofsend_each()andsend_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:
- Initialization — firebase_admin.initialize_app() with credentials and config
- Service Client — firestore.client(), auth, messaging, db (Realtime), or bucket()
- Authentication — Verify ID token or check custom claims before any user operation
- Operation — Firestore CRUD, FCM send, or Auth user management
- Error Handling — Catch firebase_admin.exceptions subclasses (UnauthenticatedError, NotFoundError, etc.)
Related Skills
| Skill | Purpose |
|---|---|
coding-supabase-sdk |
Supabase BaaS patterns (alternative to Firebase) |
coding-postgresql-sdk |
Direct PostgreSQL patterns (alternative to Firestore) |
coding-authentication-patterns |
Auth flow patterns (JWTs, sessions, OAuth) |
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