tessl/pypi-firebase-admin
Firebase Admin Python SDK enables server-side Python developers to integrate Firebase services into their applications from privileged environments.
—
Pending
realtime-database.mddocs/
Realtime Database
Firebase Realtime Database operations for JSON tree-structured data including reading, writing, querying, transactions, and real-time listeners. Provides synchronization capabilities for collaborative applications.
Capabilities
Database References
Get database references for reading and writing data at specific paths in the JSON tree structure.
def reference(path='/', app=None, url=None):
"""
Return a database reference for the specified path.
Args:
path: Database path string (default: '/')
app: Firebase app instance (optional)
url: Database URL override (optional)
Returns:
Reference: Database reference instance for the specified path
"""Data Operations
Read and write operations for JSON data with support for ETags for optimistic concurrency control.
class Reference:
"""Reference to a location in the Realtime Database."""
def get(self, etag=False, shallow=False):
"""
Get data at this reference location.
Args:
etag: Whether to return ETag for optimistic locking (optional)
shallow: Whether to return only immediate children keys (optional)
Returns:
Data at this location, or tuple of (data, etag) if etag=True
"""
def get_if_changed(self, etag):
"""
Get data only if it has changed from the provided ETag.
Args:
etag: ETag string from previous get operation
Returns:
tuple: (changed, data, new_etag) where changed is boolean
"""
def set(self, value):
"""
Set data at this reference location.
Args:
value: JSON-serializable data to set
"""
def set_if_unchanged(self, expected_etag, value):
"""
Set data only if ETag matches (optimistic locking).
Args:
expected_etag: Expected ETag string
value: JSON-serializable data to set
Returns:
tuple: (success, data, etag) where success indicates if set occurred
"""
def update(self, value):
"""
Update specific fields at this reference location.
Args:
value: Dict of field updates to apply
"""
def push(self, value=''):
"""
Push data to this reference, generating a unique child key.
Args:
value: JSON-serializable data to push (optional)
Returns:
Reference: Reference to the newly created child
"""
def delete(self):
"""Delete data at this reference location."""Navigation and Properties
Navigate the database tree structure and access reference metadata.
class Reference:
"""Reference to a location in the Realtime Database."""
@property
def key(self):
"""The key name of this reference (last segment of path)."""
@property
def path(self):
"""The full path of this reference."""
@property
def parent(self):
"""Reference to the parent location, or None if root."""
def child(self, path):
"""
Get a reference to a child location.
Args:
path: Child path string (can contain multiple segments)
Returns:
Reference: Reference to the child location
"""Query Operations
Query and order data with filtering, pagination, and sorting capabilities.
class Reference:
"""Reference to a location in the Realtime Database."""
def order_by_child(self, path):
"""
Order results by a child key.
Args:
path: Child key path to order by
Returns:
Query: Query instance for further filtering
"""
def order_by_key(self):
"""
Order results by child keys.
Returns:
Query: Query instance for further filtering
"""
def order_by_value(self):
"""
Order results by child values.
Returns:
Query: Query instance for further filtering
"""
class Query:
"""Query for filtering and ordering database results."""
def limit_to_first(self, limit):
"""
Limit results to the first N items.
Args:
limit: Maximum number of items to return
Returns:
Query: Query with limit applied
"""
def limit_to_last(self, limit):
"""
Limit results to the last N items.
Args:
limit: Maximum number of items to return
Returns:
Query: Query with limit applied
"""
def start_at(self, start):
"""
Set the starting point for results.
Args:
start: Starting value for the query
Returns:
Query: Query with start constraint
"""
def end_at(self, end):
"""
Set the ending point for results.
Args:
end: Ending value for the query
Returns:
Query: Query with end constraint
"""
def equal_to(self, value):
"""
Filter results equal to the specified value.
Args:
value: Value to match
Returns:
Query: Query with equality filter
"""
def get(self):
"""
Execute the query and get results.
Returns:
Ordered results based on query parameters
"""Real-time Listeners
Subscribe to data changes for real-time synchronization and collaborative features.
class Reference:
"""Reference to a location in the Realtime Database."""
def listen(self, callback):
"""
Start listening for data changes at this reference.
Args:
callback: Function called when data changes, receives Event instance
Returns:
ListenerRegistration: Registration object for managing the listener
"""
class ListenerRegistration:
"""Registration for a real-time database listener."""
def close(self):
"""Stop listening for changes and clean up resources."""
class Event:
"""Event object passed to listener callbacks."""
@property
def data(self):
"""The data at the reference location."""
@property
def path(self):
"""The path where the change occurred."""
@property
def event_type(self):
"""The type of change that occurred."""Transactions
Execute atomic operations on database data with automatic retry on conflicts.
class Reference:
"""Reference to a location in the Realtime Database."""
def transaction(self, transaction_update):
"""
Execute a transaction at this reference location.
Args:
transaction_update: Function that takes current data and returns new data
Returns:
The new value after the transaction completes
Raises:
TransactionAbortedError: If the transaction fails after retries
"""
class TransactionAbortedError:
"""Exception raised when a transaction is aborted."""Usage Examples
Basic Data Operations
from firebase_admin import db
# Get root reference
ref = db.reference()
# Set data
ref.child('users/user123').set({
'name': 'John Doe',
'email': 'john@example.com',
'profile': {
'age': 30,
'city': 'San Francisco'
}
})
# Get data
user_data = ref.child('users/user123').get()
print(f'User data: {user_data}')
# Update specific fields
ref.child('users/user123').update({
'profile/age': 31,
'lastLogin': datetime.datetime.now().isoformat()
})
# Push data (auto-generated key)
messages_ref = ref.child('messages')
new_message_ref = messages_ref.push({
'text': 'Hello, world!',
'timestamp': datetime.datetime.now().isoformat()
})
print(f'New message key: {new_message_ref.key}')Querying Data
# Order and limit
users_ref = ref.child('users')
query = users_ref.order_by_child('profile/age').limit_to_first(10)
young_users = query.get()
# Range queries
adult_users = (users_ref
.order_by_child('profile/age')
.start_at(18)
.end_at(65)
.get())
# Key-based ordering
recent_messages = (ref.child('messages')
.order_by_key()
.limit_to_last(20)
.get())Real-time Listeners
def on_user_change(event):
print(f'User data changed at {event.path}: {event.data}')
# Listen to specific user
user_ref = ref.child('users/user123')
listener = user_ref.listen(on_user_change)
# Listen to all users
def on_users_change(event):
print(f'Users collection changed: {event.event_type}')
users_listener = ref.child('users').listen(on_users_change)
# Stop listening
listener.close()
users_listener.close()Transactions
def increment_counter(current_value):
if current_value is None:
return 1
return current_value + 1
# Atomic increment
counter_ref = ref.child('counters/page_views')
new_count = counter_ref.transaction(increment_counter)
print(f'New count: {new_count}')
# Complex transaction
def transfer_funds(current_balances):
if current_balances is None:
return None
from_balance = current_balances.get('from', 0)
to_balance = current_balances.get('to', 0)
amount = 100
if from_balance < amount:
return None # Abort transaction
return {
'from': from_balance - amount,
'to': to_balance + amount
}
balances_ref = ref.child('balances')
try:
result = balances_ref.transaction(transfer_funds)
print(f'Transfer successful: {result}')
except db.TransactionAbortedError:
print('Transfer failed: insufficient funds')Optimistic Locking with ETags
# Get data with ETag
data, etag = ref.child('settings').get(etag=True)
# Modify data
data['last_updated'] = datetime.datetime.now().isoformat()
# Set only if unchanged
try:
success, new_data, new_etag = ref.child('settings').set_if_unchanged(etag, data)
if success:
print('Settings updated successfully')
else:
print('Settings were modified by another process')
except Exception as e:
print(f'Update failed: {e}')Database URL Configuration
# Use specific database URL
custom_ref = db.reference('/', url='https://my-other-project-default-rtdb.firebaseio.com/')
# Or configure in app initialization
import firebase_admin
from firebase_admin import credentials
cred = credentials.Certificate('serviceAccountKey.json')
app = firebase_admin.initialize_app(cred, {
'databaseURL': 'https://my-project-default-rtdb.firebaseio.com/'
})
# Then use default reference
ref = db.reference()Security and Performance
Security Rules
Realtime Database security is managed through security rules defined in the Firebase Console:
// Example security rules (defined in Firebase Console)
{
"rules": {
"users": {
"$uid": {
".read": "$uid === auth.uid",
".write": "$uid === auth.uid"
}
},
"public": {
".read": true,
".write": false
}
}
}Performance Tips
- Denormalize data: Structure data to minimize reads
- Use shallow queries: Get only keys when possible
- Limit listener scope: Listen to specific paths, not large trees
- Index data properly: Structure keys for efficient queries
- Use transactions for atomic operations: Ensure data consistency
- Close listeners: Always clean up real-time listeners when done
Types
class Reference:
"""Reference to a location in the Realtime Database."""
@property
def key(self):
"""The key name of this reference."""
@property
def path(self):
"""The full path of this reference."""
@property
def parent(self):
"""Reference to the parent location."""
class Query:
"""Query for filtering and ordering database results."""
class Event:
"""Event object for real-time listeners."""
@property
def data(self):
"""The data at the reference location."""
@property
def path(self):
"""The path where the change occurred."""
@property
def event_type(self):
"""The type of change that occurred."""
class ListenerRegistration:
"""Registration for a real-time database listener."""
class TransactionAbortedError(Exception):
"""Exception raised when a transaction is aborted."""Install with Tessl CLI
npx tessl i tessl/pypi-firebase-admin