Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def publish_py(message, channel=None, event_class="default", data=None,
site=SITE_NAME, target=None):
cent_url = CENTRIFUGO_HOST + ":" + str(CENTRIFUGO_PORT)
client = Client(cent_url, SECRET_KEY, timeout=1)
channel = _get_channel(channel, target)
if data is None:
data = {}
payload = {"message": message, "channel": channel,
'event_class': event_class, "data": data, "site": site}
err = None
try:
client.publish(channel, payload)
except CentException as e:
err = str(e)
if event_class.lower() == "debug":
print("[DEBUG] ", str(json.dumps(payload)))
return err