Files
app-autotest/site/operations/tracker.py
T

31 lines
1.1 KiB
Python

"""Трекер созданных инстансов — in-memory dict (один gunicorn worker)."""
import threading
_LOCK = threading.Lock()
_data = {
"408b7f96-6ed7-4985-be1b-f5bdcb6ab44d": {"svcId": 1, "displayName": "autotest-1", "instanceUid": "408b7f96-6ed7-4985-be1b-f5bdcb6ab44d"},
"884356cf-5212-4395-b56b-27c58a5fc1fa": {"svcId": 1, "displayName": "autotest-1-first", "instanceUid": "884356cf-5212-4395-b56b-27c58a5fc1fa"},
"6528854d-a4ea-428c-9fa4-68e85fa9b3ac": {"svcId": 1, "displayName": "curl-test-144834", "instanceUid": "6528854d-a4ea-428c-9fa4-68e85fa9b3ac"},
"fdcb5887-39f7-4ef8-a9c0-d55e434a55ff": {"svcId": 1, "displayName": "curl-final-test", "instanceUid": "fdcb5887-39f7-4ef8-a9c0-d55e434a55ff"},
}
def add(instance_uid, svc_id, display_name):
with _LOCK:
_data[instance_uid] = {
"svcId": svc_id,
"displayName": display_name,
"instanceUid": instance_uid,
}
def remove(instance_uid):
with _LOCK:
_data.pop(instance_uid, None)
def list_all():
with _LOCK:
return list(_data.values())