I'm reading data from an Update Cloud Firestore Trigger. The event is a dictionary that contains the data whithin the key ['value']['fields']. However, each of the keys contains s nested dictionary containing a key like 'integerValue', 'booleanValue' or 'stringValue', where the value of integerValue is actually a string. Is there a method to remove the 'type pointers'?
How can I convert this:
{
'fields': {
'count': {
'integerValue': '0'
},
'verified': {
'booleanValue': False
},
'user': {
'stringValue': 'Matt'
}
}
}
To this:
{
'count': 0,
'verified': False,
'user': 'Matt',
}
Recently i encountered similar problem.
We could recursively traverse the map to extract and simplify the event trigger data.
Here's python implementation, extended from previous answers.
class FirestoreTriggerConverter(object):
def __init__(self, client=None) -> None:
self.client = client if client else firestore.client()
self._action_dict = {
'geoPointValue': (lambda x: dict(x)),
'stringValue': (lambda x: str(x)),
'arrayValue': (lambda x: [self._parse_value(value_dict) for value_dict in x.get("values", [])]),
'booleanValue': (lambda x: bool(x)),
'nullValue': (lambda x: None),
'timestampValue': (lambda x: self._parse_timestamp(x)),
'referenceValue': (lambda x: self._parse_doc_ref(x)),
'mapValue': (lambda x: {key: self._parse_value(value) for key, value in x["fields"].items()}),
'integerValue': (lambda x: int(x)),
'doubleValue': (lambda x: float(x)),
}
def convert(self, data_dict: dict) -> dict:
result_dict = {}
for key, value_dict in data_dict.items():
result_dict[key] = self._parse_value(value_dict)
return result_dict
def _parse_value(self, value_dict: dict) -> Any:
data_type, value = value_dict.popitem()
return self._action_dict[data_type](value)
def _parse_timestamp(self, timestamp: str):
try:
return datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%S.%fZ')
except ValueError as e:
return datetime.strptime(timestamp, '%Y-%m-%dT%H:%M:%SZ')
def _parse_doc_ref(self, doc_ref: str) -> DocumentReference:
path_parts = doc_ref.split('/documents/')[1].split('/')
collection_path = path_parts[0]
document_path = '/'.join(path_parts[1:])
doc_ref = self.client.collection(collection_path).document(document_path)
return doc_ref
Use this as follows
converter = FirestoreTriggerConverter(client)
simplified_data_dict = converter.convert(event_data_dict["event"]["value"]["fields"])
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With