A type-safe wrapper around Google Cloud Firestore for internal use. This library provides a clean, Pydantic-based interface for Firestore operations with full type safety and modern Python features.
- 🔒 Type-safe queries with compile-time field validation
- 📝 Pydantic integration for automatic serialization/deserialization
- 🛠️ Generic repository pattern for consistent CRUD operations
- 🔍 Fluent query builder with method chaining
- 📊 Automatic timestamping for created_at and updated_at fields
- 🎯 Field reference system for IDE autocompletion and type checking
uv add git+https://github.com/InfinityAILab/infinity-firestore-python --tag v0.1.0- Python 3.9+
- Google Cloud Firestore
- Pydantic v2
- Firebase Admin SDK
from infinity_firestore import initialize_firebase
# Initialize Firebase (typically done once at app startup)
initialize_firebase()from infinity_firestore.model import Model
from pydantic import Field
class User(Model):
name: str
email: str
age: int = Field(gt=0)
is_active: bool = Truefrom infinity_firestore.repository import FirestoreRepository
class UserRepository(FirestoreRepository[User]):
def __init__(self):
super().__init__("users", User)
# Create repository instance
user_repo = UserRepository()# Create a new user
user = User(name="John Doe", email="[email protected]", age=30)
created_user = await user_repo.create(user)
# Get user by ID
user = await user_repo.get_by_id(created_user.id)
# Update user
user.age = 31
updated_user = await user_repo.update(user.id, user)
# Delete user
await user_repo.delete(user.id)
# List all users
users = await user_repo.list_all(limit=10)The library provides a type-safe query builder that prevents runtime errors:
# Get field references for type-safe queries
fields = user_repo.fields()
# Build and execute type-safe queries
query = FirestoreQueryBuilder(user_repo.collection, fields) \
.where(fields.age, ">=", 18) \
.where(fields.is_active, "==", True) \
.order_by(fields.created_at, "desc") \
.limit(10)
adults = await user_repo.query(query)# Find by single field
active_users = await user_repo.find_by_field("is_active", True)
# Find by multiple fields
john_users = await user_repo.find_by_fields(
name="John Doe",
is_active=True
)from infinity_firestore import get_firestore_client
# Use a specific database
client = get_firestore_client(database="my-custom-db")
class UserRepository(FirestoreRepository[User]):
def __init__(self):
super().__init__("users", User)
self._db = client # Override default clientInitializes the Firebase Admin SDK. Should be called once at application startup.
Returns a Firestore client instance.
Parameters:
database: Optional database name. If None, uses the default database.
Base class for all Firestore models. Provides automatic ID generation and timestamps.
Fields:
id: str- Auto-generated document IDcreated_at: datetime- UTC timestamp when document was createdupdated_at: datetime- UTC timestamp when document was last updated
Generic repository for CRUD operations on Firestore collections.
Constructor:
def __init__(self, collection_name: str, model_class: Type[T])Methods:
Creates a new document in Firestore.
Retrieves a document by its ID. Returns None if not found.
Updates an existing document. Automatically updates the updated_at timestamp.
Deletes a document by ID. Returns True if successful.
Lists all documents in the collection with optional limit.
Finds documents where a field equals a specific value.
Finds documents matching multiple field criteria.
Executes a query built with FirestoreQueryBuilder.
Returns typed field references for the model.
Fluent builder for constructing type-safe Firestore queries.
Methods:
Adds a where clause to the query.
Supported operators:
"=="- Equal"!="- Not equal"<"- Less than"<="- Less than or equal">"- Greater than">="- Greater than or equal"in"- In array"not-in"- Not in array"array-contains"- Array contains value"array-contains-any"- Array contains any of the values
Orders results by a field.
Directions:
Query.ASCENDING(default)Query.DESCENDING
Limits the number of results.
Returns the built Firestore Query object.
Executes the query and returns results as a list of models.
The library uses Python's standard exception handling. All repository methods may raise:
Exception- For general Firestore operation failuresTypeError- For type mismatches in queriesAttributeError- For invalid field references
try:
user = await user_repo.get_by_id("invalid-id")
except Exception as e:
logger.error(f"Failed to get user: {e}")Always use proper type hints for better IDE support and runtime safety:
from typing import Optional
async def get_user_by_email(email: str) -> Optional[User]:
users = await user_repo.find_by_field("email", email)
return users[0] if users else NoneWrap repository calls in try-catch blocks:
async def safe_create_user(user_data: dict) -> Optional[User]:
try:
user = User(**user_data)
return await user_repo.create(user)
except Exception as e:
logger.error(f"Failed to create user: {e}")
return NoneAlways use the type-safe field references:
# ✅ Good - Type-safe
fields = user_repo.fields()
query = FirestoreQueryBuilder(user_repo.collection, fields) \
.where(fields.age, ">=", 18)
# ❌ Bad - String-based, error-prone
await user_repo.find_by_field("age", 18) # Typos not caught at compile timeInitialize Firebase at application startup:
# main.py or app initialization
from infinity_firestore import initialize_firebase
def setup_app():
initialize_firebase()
# ... rest of app setup- This library is designed for internal use and may not handle all edge cases
- No built-in pagination beyond simple
limit() - No support for Firestore transactions or batch operations
- Limited support for complex nested queries
- No built-in caching mechanisms
This is an internal library and is not maintained as an open-source project. Use at your own risk.