danieldanciu

    logging.info(f'Watching Firestore collection {firestore_prefix}users/<user_id>/jobs '
                 f'for documents with status=={statuses_of_interest}')
    sync_client = firestore.Client(credentials=creds, project=project_id)
    # a Firebase index on collection_id and field status MUST exist for this query to work
    watched_jobs = sync_client.collection_group(job_collection_id).where('status', 'in', statuses_of_interest)

    ...
    
    watched_jobs.on_snapshot(callback)

danieldanciu

def callback(docs: List[firestore.DocumentSnapshot],
                 changes: List[DocumentChange],
                 _: DatetimeWithNanoseconds):
        try:
            for document, change in zip(docs, changes):
                if change.type in {ChangeType.ADDED, ChangeType.MODIFIED}:
                    job_doc = firestore.DocumentReference(*document.reference._path, client=db_client)
                    if not job_doc.path.startswith(f'{firestore_prefix}users'):
                        continue
                    pre_status = document.get('status')