Class GCSBackend

celery/backends/gcs.py:161–354

Google Cloud Storage task result backend. Uses Firestore for chord ref count.

class GCSBackend(GCSBackendBase):
    """Google Cloud Storage task result backend.

    Uses Firestore for chord ref count.
    """

    implements_incr = True
    supports_native_join = True

    # Firestore parameters
    _collection_name = 'celery'
    _field_count = 'chord_count'
    _field_expires = 'expires_at'

    def __init__(self, **kwargs):
        if not (firestore and firestore_admin_v1):
            raise ImproperlyConfigured(
                'You must install google-cloud-firestore to use gcs backend'
            )
        super().__init__(**kwargs)

        self._firestore_lock = RLock()
        self._firestore_client = None

        self.firestore_project = self.app.conf.get(
            'firestore_project', self.project
        )
        if not self._is_firestore_ttl_policy_enabled():
            raise ImproperlyConfigured(
                f'Missing TTL policy to use gcs backend with ttl on '
                f'Firestore collection: {self._collection_name} '
                f'project: {self.firestore_project}'
            )

    @property
    def firestore_client(self):
        """Returns a firestore client."""

        # make sure it's thread-safe, as creating a new client is expensive
        with self._firestore_lock:
            if self._firestore_client and self._pid == getpid():
                return self._firestore_client
            # make sure each process gets its own connection after a fork
            self._firestore_client = firestore.Client(
                project=self.firestore_project
            )
            self._pid = getpid()
        return self._firestore_client

    def _is_firestore_ttl_policy_enabled(self):
        client = firestore_admin_v1.FirestoreAdminClient()

        name = (
            f"projects/{self.firestore_project}"
            f"/databases/(default)/collectionGroups/{self._collection_name}"
            f"/fields/{self._field_expires}"
        )
        request = firestore_admin_v1.GetFieldRequest(name=name)