moved stuff around
This commit is contained in:
parent
d8bdd717c7
commit
8668550a3f
62
app.py
62
app.py
@ -1,14 +1,15 @@
|
|||||||
#!/usr/bin/env python
|
#!/usr/bin/env python3
|
||||||
import sentry_sdk
|
import sentry_sdk
|
||||||
import time
|
import time
|
||||||
import requests
|
import requests
|
||||||
import requests.exceptions
|
import requests.exceptions
|
||||||
import os
|
import os
|
||||||
import redis
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
import socket
|
import socket
|
||||||
|
|
||||||
|
from redis_super_storage import RedisSuperStorage
|
||||||
|
|
||||||
"""
|
"""
|
||||||
Scheduler
|
Scheduler
|
||||||
"""
|
"""
|
||||||
@ -21,63 +22,6 @@ __version__text__ = "1"
|
|||||||
sentry_sdk.init("https://0a106e104e114bc9a3fa47f9cb0db2f4@sentry.kmlabz.com/10")
|
sentry_sdk.init("https://0a106e104e114bc9a3fa47f9cb0db2f4@sentry.kmlabz.com/10")
|
||||||
|
|
||||||
|
|
||||||
class RedisSuperStorage:
|
|
||||||
def __init__(self, redis_url: str):
|
|
||||||
self.r = redis.from_url(redis_url)
|
|
||||||
|
|
||||||
def get_consumer_list(self) -> dict:
|
|
||||||
keys = self.r.keys('consumer_*')
|
|
||||||
|
|
||||||
list_of_customers = {}
|
|
||||||
|
|
||||||
for key in keys:
|
|
||||||
info = json.loads((self.r.get(key) or b"{}").decode('utf-8'))
|
|
||||||
|
|
||||||
if info:
|
|
||||||
list_of_customers[info['uuid']] = info
|
|
||||||
|
|
||||||
return list_of_customers
|
|
||||||
|
|
||||||
def get_producer_list(self) -> list:
|
|
||||||
keys = self.r.keys('producer_*')
|
|
||||||
|
|
||||||
list_of_producer_ip = []
|
|
||||||
|
|
||||||
for key in keys:
|
|
||||||
ip = (self.r.get(key) or b"").decode('utf-8')
|
|
||||||
|
|
||||||
if ip:
|
|
||||||
list_of_producer_ip.append(ip)
|
|
||||||
|
|
||||||
return list_of_producer_ip
|
|
||||||
|
|
||||||
def is_ip_changed(self) -> bool:
|
|
||||||
old_ip = self.r.get('current_ip')
|
|
||||||
|
|
||||||
if old_ip:
|
|
||||||
old_ip = old_ip.decode('utf-8')
|
|
||||||
|
|
||||||
host_name = socket.gethostname()
|
|
||||||
current_ip = socket.gethostbyname(host_name)
|
|
||||||
|
|
||||||
if current_ip != old_ip:
|
|
||||||
self.r.set('current_ip', current_ip.encode('utf-8'))
|
|
||||||
|
|
||||||
return current_ip != old_ip
|
|
||||||
|
|
||||||
def update_consumer(self, uuid: str, ip: str):
|
|
||||||
|
|
||||||
cust_key = f"consumer_{uuid}"
|
|
||||||
|
|
||||||
info = {
|
|
||||||
"uuid": uuid,
|
|
||||||
"ip": ip,
|
|
||||||
"last_seen": time.time()
|
|
||||||
}
|
|
||||||
|
|
||||||
self.r.set(cust_key, json.dumps(info).encode('utf-8'))
|
|
||||||
self.r.expire(cust_key, os.environ["CUSTOMER_TIMEOUT"])
|
|
||||||
|
|
||||||
|
|
||||||
class Scheduler:
|
class Scheduler:
|
||||||
|
|
||||||
|
65
redis_super_storage.py
Normal file
65
redis_super_storage.py
Normal file
@ -0,0 +1,65 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import redis
|
||||||
|
import os
|
||||||
|
import json
|
||||||
|
import socket
|
||||||
|
import time
|
||||||
|
|
||||||
|
|
||||||
|
class RedisSuperStorage:
|
||||||
|
def __init__(self, redis_url: str, timeout: int):
|
||||||
|
self.r = redis.from_url(redis_url)
|
||||||
|
self._timeout = timeout
|
||||||
|
|
||||||
|
def get_consumer_list(self) -> dict:
|
||||||
|
keys = self.r.keys('consumer_*')
|
||||||
|
|
||||||
|
list_of_customers = {}
|
||||||
|
|
||||||
|
for key in keys:
|
||||||
|
info = json.loads((self.r.get(key) or b"{}").decode('utf-8'))
|
||||||
|
|
||||||
|
if info:
|
||||||
|
list_of_customers[info['uuid']] = info
|
||||||
|
|
||||||
|
return list_of_customers
|
||||||
|
|
||||||
|
def get_producer_list(self) -> list:
|
||||||
|
keys = self.r.keys('producer_*')
|
||||||
|
|
||||||
|
list_of_producer_ip = []
|
||||||
|
|
||||||
|
for key in keys:
|
||||||
|
ip = (self.r.get(key) or b"").decode('utf-8')
|
||||||
|
|
||||||
|
if ip:
|
||||||
|
list_of_producer_ip.append(ip)
|
||||||
|
|
||||||
|
return list_of_producer_ip
|
||||||
|
|
||||||
|
def is_ip_changed(self) -> bool:
|
||||||
|
old_ip = self.r.get('current_ip')
|
||||||
|
|
||||||
|
if old_ip:
|
||||||
|
old_ip = old_ip.decode('utf-8')
|
||||||
|
|
||||||
|
host_name = socket.gethostname()
|
||||||
|
current_ip = socket.gethostbyname(host_name)
|
||||||
|
|
||||||
|
if current_ip != old_ip:
|
||||||
|
self.r.set('current_ip', current_ip.encode('utf-8'))
|
||||||
|
|
||||||
|
return current_ip != old_ip
|
||||||
|
|
||||||
|
def update_consumer(self, uuid: str, ip: str):
|
||||||
|
|
||||||
|
cust_key = f"consumer_{uuid}"
|
||||||
|
|
||||||
|
info = {
|
||||||
|
"uuid": uuid,
|
||||||
|
"ip": ip,
|
||||||
|
"last_seen": time.time()
|
||||||
|
}
|
||||||
|
|
||||||
|
self.r.set(cust_key, json.dumps(info).encode('utf-8'))
|
||||||
|
self.r.expire(cust_key, os.environ["CUSTOMER_TIMEOUT"])
|
Reference in New Issue
Block a user