Decorators with arguments?
Christopher de Vidal
cbdevidal.jk1 at gmail.com
Thu May 14 08:28:23 EDT 2020
Help please? Creating an MQTT-to-Firestore bridge and I know a decorator
would help but I'm stumped how to create one. I've used decorators before
but not with arguments.
The Firestore collection.on_snapshot() method invokes a callback and sends
it three parameters (collection_snapshot, changes, and read_time). I need
the callback to also know the name of the collection so that I can publish
to the equivalent MQTT topic name. I had thought to add a fourth parameter
and I believe a decorator is the right approach but am stumped how to add
that fourth parameter. How would I do this with the code below?
#!/usr/bin/env python3
from google.cloud import firestore
import firebase_admin
from firebase_admin import credentials
import json
import mqtt
firebase_admin.initialize_app(credentials.Certificate("certs/firebase.json"))
db = firestore.Client()
mqtt.connect()
def load_json(contents):
try:
return json.loads(contents)
except (json.decoder.JSONDecodeError, TypeError):
return contents
def on_snapshot(col_name, col_snapshot, changes, read_time):
data = dict()
for doc in col_snapshot:
serial = doc.id
contents = load_json(doc.to_dict()['value'])
data[serial] = contents
for change in changes:
serial = change.document.id
mqtt_topic = col_name + '/' + serial
contents = data[serial]
if change.type.name in ['ADDED', 'MODIFIED']:
mqtt.publish(mqtt_topic, contents)
elif change.type.name == 'REMOVED':
mqtt.publish(mqtt_topic, None)
# Start repeated code section
# TODO Better to use decorators but I was stumped on how to pass arguments
def door_status_on_snapshot(col_snapshot, changes, read_time):
on_snapshot('door_status', col_snapshot, changes, read_time)
door_status_col_ref = db.collection('door_status')
door_status_col_watch =
door_status_col_ref.on_snapshot(door_status_on_snapshot)
# Repetition...
def cpu_temp_on_snapshot(col_snapshot, changes, read_time):
on_snapshot('cpu_temp', col_snapshot, changes, read_time)
cpu_temp_col_ref = db.collection('cpu_temp')
cpu_temp_col_watch = cpu_temp_col_ref.on_snapshot(cpu_temp_on_snapshot)
# End repeated code section
# Start repeated code section
door_status_col_watch.unsubscribe()
cpu_temp_col_watch.unsubscribe()
# Repetition...
# End repeated code section
Christopher de Vidal
Would you consider yourself a good person? Have you ever taken the 'Good
Person' test? It's a fascinating five minute quiz. Google it.
More information about the Python-list
mailing list