Cloud Functions Listen to Firestore Map Field Changes (Python)

Refer to Cloud Functions Listen to Firestore Triggers for setup, deployment and observe normal fields.

This article focus on detecting changes to map fields named likes with the following structure.

Observe create/update/delete of likes
Observe addition and removal from likes
Assume likes[KEY] will not be modified upon creation, but might be deleted

likes:
    key01: value
    key02: value

import logginglog = logging.getLogger(__name__)def observe_user_activity(data, context):    if not data['oldValue'] and data['value']:        log.info('create document')        if 'likes' in data['value']['fields']:            likes = data['value']['fields']['likes']['mapValue']['fields']            for key, value in likes.items():                log.info(f"{key}={value}")    elif data['oldValue'] and not data['value']:        log.info('delete document')        if 'likes' in data['oldValue']['fields']:            likes = data['oldValue']['fields']['likes']['mapValue']['fields']            for key, value in likes.items():                log.info(f"{key}={value}")    elif data['updateMask']:        for _field in data['updateMask']['fieldPaths']:            if _field == 'likes':                if _field not in data['oldValue']['fields'] and _field in data['value']['fields']:                    log.info(f'new field: {_field}')                    if _field == 'likes':                        likes = data['value']['fields']['likes']['mapValue']['fields']                        for key, value in likes.items():                            log.info(f"{key}={value}")                elif _field in data['oldValue']['fields'] and _field not in data['value']['fields']:                    log.info(f'delete field: {_field}')                    if _field == 'likes':                        likes = data['oldValue']['fields']['likes']['mapValue']['fields']                        for key, value in likes.items():                            log.info(f"{key}={value}")                else:                    log.info(f'modified field: {_field}')            elif _field.startswith('likes.'):                key = _field[6:]                exist_in_old_value = True                try:                    old_value = data['oldValue']['fields']['likes']['mapValue']['fields'][key]                except KeyError:                    exist_in_old_value = False                exist_in_value = True                try:                    value = data['value']['fields']['likes']['mapValue']['fields'][key]                except KeyError:                    exist_in_value = False                if not exist_in_old_value and exist_in_value:                    log.info(f'new field: {_field}')                     log.info(f"{key}={value}")                elif exist_in_old_value and not exist_in_value:                    log.info(f'delete field: {_field}')                     log.info(f"{key}={old_value}")                else:                    log.info(f'modified field: {_field}')            else:                log.info(f'modified field: {_field}')