Praudyog

EAP – GCP Firestore Storage

admin

EAP – GCP Firestore Storage

Example:

Import Firestore:

from google.cloud import firestore

Create Firestore Class:

class FirestoreAdapter(DatabaseAdapter):
    """Firestore database adapter implementation."""

    COLLECTION_INTEGRATIONS = "integrations"
    COLLECTION_JOBS = "jobs"
    COLLECTION_FILE_ITEMS = "file_items"
    COLLECTION_ERRORS = "errors"
    COLLECTION_TENANTS = "tenants"

    def __init__(self):
        """Initialize Firestore client."""
        self.db: Optional[firestore.AsyncClient] = None
        self.project_id = settings.gcp_project_id
        self.database_id = settings.FIRESTORE_DATABASE_ID

Establish Firestore Connection:

async def connect(self) -> None:
        """Establish Firestore connection."""
        try:
            self.db = firestore.AsyncClient(
                project=self.project_id,
                database=self.database_id,
                credentials=get_gcp_credentials(),
            )
            logger.info(f"Connected to Firestore: {self.project_id}/{self.database_id}")
        except Exception as e:
            # Log only the exception type — the message may contain GCP project
            # details or auth tokens that should not appear in logs.
            logger.error(
                "Failed to connect to Firestore [%s]: %s",
                type(e).__name__,
                # Include only the first 120 chars; avoid dumping raw credentials
                # that GCP client libraries sometimes embed in error strings.
                str(e)[:120] if str(e) else "(no message)",
            )
            raise

Disconnect Firestore Connection:

async def disconnect(self) -> None:
        """Close Firestore connection."""
        if self.db:
            self.db.close()
            logger.info("Disconnected from Firestore")

Create Firestore Collection & Save a Document:

# Integration operations
    async def create_integration(self, integration: Integration) -> Integration:
        """Create a new integration."""
        doc_ref = self.db.collection(self.COLLECTION_INTEGRATIONS).document(
            integration.integration_id
        )
        await doc_ref.set(integration.to_dict())
        logger.info(f"Created integration: {integration.integration_id}")
        return integration

Fetch Document By Id:


    async def get_integration(self, integration_id: str) -> Optional[Integration]:
        """Get integration by ID."""
        doc_ref = self.db.collection(self.COLLECTION_INTEGRATIONS).document(integration_id)
        doc = await doc_ref.get()

        if doc.exists:
            data = doc.to_dict()
            return integration_from_dict(data)
        return None

Update A Document From Firestore Collection:

async def update_integration(self, integration: Integration) -> Integration:
        """Update existing integration."""
        doc_ref = self.db.collection(self.COLLECTION_INTEGRATIONS).document(
            integration.integration_id
        )
        await doc_ref.update(integration.to_dict())
        logger.info(f"Updated integration: {integration.integration_id}")
        return integration

List All Documents From A Firestore Collection:

async def list_integrations(self, is_active: bool = True) -> List[Integration]:
        """List all integrations."""
        query = self.db.collection(self.COLLECTION_INTEGRATIONS).where(
            filter=FieldFilter("is_active", "==", is_active)
        )

        docs = query.stream()
        integrations = []
        async for doc in docs:
            data = doc.to_dict()
            integrations.append(integration_from_dict(data))

        return integrations

List All Documents From A Firestore Collection By Tanent Id:

async def list_integrations_by_tenant(
        self,
        tenant_id: str,
        source: Optional[str] = None,
        is_active: bool = True,
    ) -> List[Integration]:
        """List integrations scoped to a platform tenant, with optional source filter.

        Uses a single equality filter on platform_tenant_id (auto-indexed) and
        applies is_active / source filtering in Python to avoid composite indexes.
        """
        query = self.db.collection(self.COLLECTION_INTEGRATIONS).where(
            filter=FieldFilter("platform_tenant_id", "==", tenant_id)
        )

        integrations: List[Integration] = []
        async for doc in query.stream():
            intg = integration_from_dict(doc.to_dict())
            if intg.is_active != is_active:
                continue
            if source and intg.source != source:
                continue
            integrations.append(intg)
        return integrations

Create A Job Collection & Save The Job Document:

async def create_job(self, job: Job) -> Job:
        """Create a new job."""
        doc_ref = self.db.collection(self.COLLECTION_JOBS).document(job.job_id)
        await doc_ref.set(job.to_dict())
        logger.info(f"Created job: {job.job_id}")
        return job

Create The Job Collection By Its Id:

    async def get_job(self, job_id: str) -> Optional[Job]:
        """Get job by ID."""
        doc_ref = self.db.collection(self.COLLECTION_JOBS).document(job_id)
        doc = await doc_ref.get()

        if doc.exists:
            data = doc.to_dict()
            return job_from_dict(data)
        return None

Update An Existing Job Collection Document By Its Id


    async def update_job(self, job_id: str, fields: dict) -> None:
        """Update existing job fields by job ID."""
        doc_ref = self.db.collection(self.COLLECTION_JOBS).document(job_id)
        await doc_ref.update(fields)

Leave a Reply

Your email address will not be published. Required fields are marked *