Simplifying Firestore Interactions with FSEntity Class

Introduction

Managing data in Google Firestore can involve repetitive tasks like creating, retrieving, updating, and deleting documents. While Firestore offers a powerful API, writing boilerplate code for these common operations can be time-consuming and error-prone. The FSEntity class comes to the rescue, providing a streamlined approach to interacting with Firestore in Python.

FSEntity Class In Depth

# firestore_client.py
import firebase_admin
from firebase_admin import credentials, firestore
from consts import PROJECT_ID, CREDENTIALS_PATH


def _FirestoreClient():
    firebase_admin.initialize_app(
        credentials.Certificate(CREDENTIALS_PATH), {"projectId": PROJECT_ID}
    )
    return firestore.client()


firestore_client = _FirestoreClient()

# fsentity.py
from abc import ABC, abstractmethod
from firestore_client import firestore_client


class FSEntity(ABC):
    # This is a generic class that can be used to interact with Firestore.
    # This class should be inherited by all the entities that are stored in Firestore.
    # This class provides the basic CRUD operations.
    # This class is not meant to be used directly. It should be inherited by the entities.
    # Anytime, you want a new class to be uploaded to Firestore,
    # 1. inherit this class
    # 2. implement to_dict
    # 3. implement from_dict
    # 4. implement key property in __post_init__ function
    db = firestore_client
    collection_name = ""  # This will be reflected in the Firestore. Example: f"{MACHINE_NAME}-employees"
    key = ""  # Override this in the child class in the __post_init__ function.

    @abstractmethod
    def to_dict(self):
        raise NotImplementedError

    @staticmethod
    @abstractmethod
    def from_dict(fsDict):
        raise NotImplementedError

    @classmethod
    def filter(cls, conditions):
        # conditions is a list of tuples. Each tuple is of the form (field, operator, value)
        # Example: conditions = [("department", "==", "Accounting"), ("monthly_salary", ">", 10000)]
        query = cls.db.collection(cls.collection_name)
        for field, operator, value in conditions:
            query = query.where(field, operator, value)
        dataObjs = query.get()
        return [cls.from_dict(data.to_dict()) for data in dataObjs]

    @classmethod
    def get_all_entities(cls):
        dataObjs = cls.db.collection(cls.collection_name).get()
        return [cls.from_dict(data.to_dict()) for data in dataObjs]

    @classmethod
    def does_entity_exist(cls, key: str):
        data = cls.db.collection(cls.collection_name).document(key).get()
        return data.exists

    @classmethod
    def get_entity(cls, key: str):
        data = cls.db.collection(cls.collection_name).document(key).get()
        # Note: Each entity implements a from_dict and to_dict function.
        if not data.exists:
            raise ValueError(f"{cls.__name__} with id {key} does not exist.")
        return cls.from_dict(data.to_dict())

    @classmethod
    def create_entity(cls, entityObj):
        entity_data = entityObj.to_dict()
        return (
            cls.db.collection(cls.collection_name)
            .document(entityObj.key)
            .set(entity_data)
        )

    @classmethod
    def update_entity(cls, entityObj):
        entity_data = entityObj.to_dict()
        return (
            cls.db.collection(cls.collection_name)
            .document(entityObj.key)
            .update(entity_data)
        )

    @classmethod
    def delete_entity(cls, key: str):
        return cls.db.collection(cls.collection_name).document(key).delete()

    @classmethod
    def delete_all_entities(cls):
        dataObjs = cls.db.collection(cls.collection_name).get()
        for data in dataObjs:
            data.reference.delete()
        return

How to Use FSEntity Class

The FSEntity class serves as a base class for your Firestore entity classes. Here’s a step-by-step guide on how to leverage its functionalities:

Create an Entity Class:

Inherit from the FSEntity class to create a new class representing your Firestore entity (e.g., TODO in the example). 2. Define to_dict and from_dict methods:

Implement the to_dict method to convert your entity object into a dictionary format suitable for Firestore storage.
Implement the from_dict method (as a static method) to convert a Firestore document dictionary back to your entity object.

Set collection_name (Optional):

If your entity belongs to a specific collection in Firestore, define the collection name within your entity class (e.g., collection_name = "TODO").

Utilize CRUD operations:

The FSEntity class provides pre-built methods for interacting with your entities:

get_all_entities: Retrieves all entities from the collection.
get_entity(key): Retrieves a specific entity by its key.
create_entity(entityObj): Creates a new entity in Firestore.
update_entity(entityObj): Updates an existing entity in Firestore.
delete_entity(key): Deletes an entity from Firestore.
does_entity_exist(key): Checks if an entity exists in Firestore based on its key.
filter(conditions): Retrieves entities based on specified filtering conditions (refer to test_fsentity.py for an example).

Example Usage:

The provided test_fsentity.py demonstrates how to create a TODO entity class inheriting from FSEntity. It showcases various functionalities like creating and retrieving TODO items, updating titles, and deleting entities.

# test_fsentity.py
from pprint import pprint as pp
from fsentity import FSEntity
from dataclasses import dataclass


@dataclass
class TODO(FSEntity):
    key: str = ""
    title: str = ""
    collection_name = "TODO"

    def __post_init__(self):
        if self.key == "":
            self.key = self.title

    def to_dict(self):
        return {"title": self.title}

    @staticmethod
    def from_dict(fsDict):
        return TODO(key=fsDict["title"], title=fsDict["title"])


def cleanup():
    TODO.delete_all_entities()
    return


def test_fsentity():
    cleanup()
    todo1 = TODO(key="1", title="Buy groceries")
    todo2 = TODO(key="2", title="Do laundry")

    all_todos = TODO.get_all_entities()
    pp(all_todos)
    assert len(all_todos) == 0

    TODO.create_entity(todo1)
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 1

    fetched_todo1 = TODO.get_entity(todo1.key)
    assert fetched_todo1.title == "Buy groceries"

    does_todo1_exists = TODO.does_entity_exist(todo1.key)
    assert does_todo1_exists == True

    TODO.create_entity(todo2)
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 2

    todo1.title = "Buy groceries and milk"
    TODO.update_entity(todo1)
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 2
    assert all_todos[0].title == "Buy groceries and milk"

    todo3 = TODO(key="3", title="Buy groceries and milk")
    TODO.create_entity(todo3)
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 3
    assert all_todos[0].title == "Buy groceries and milk"
    assert all_todos[1].title == "Do laundry"
    assert all_todos[2].title == "Buy groceries and milk"

    filtered_entities = TODO.filter([("title", "==", "Buy groceries and milk")])
    assert len(filtered_entities) == 2

    TODO.delete_entity(todo1.key)
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 2

    pp(all_todos)
    TODO.delete_all_entities()
    all_todos = TODO.get_all_entities()
    assert len(all_todos) == 0

    return

Conclusion

The FSEntity class acts as a valuable abstraction layer, reducing boilerplate code and simplifying Firestore interactions in Python. By inheriting from FSEntity and implementing the required methods, you can create custom entity classes that seamlessly manage data within your Firestore database. This not only saves development time but also promotes code maintainability and consistency within your project.