#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ¯\_(ツ)_/¯
#
# File: packages/ubdcc-mgmt/ubdcc_mgmt/Database.py
#
# Project website: https://github.com/oliver-zehentleitner/unicorn-binance-depth-cache-cluster
# Github: https://github.com/oliver-zehentleitner/unicorn-binance-depth-cache-cluster
# Documentation: https://oliver-zehentleitner.github.io/unicorn-binance-depth-cache-cluster
# PyPI: https://pypi.org/project/ubdcc-shared-modules
#
# License: MIT
# https://github.com/oliver-zehentleitner/unicorn-binance-depth-cache-cluster/blob/master/LICENSE
#
# Author: Oliver Zehentleitner
#
# Copyright (c) 2024-2026, Oliver Zehentleitner (https://about.me/oliver-zehentleitner)
# All rights reserved.
import threading
import time
import uuid
[docs]
class Database:
def __init__(self, app=None):
self.app = app
self.app.data["db"] = self
self.data = {}
self.data_lock = threading.Lock()
self._init()
def _init(self) -> bool:
self.app.stdout_msg(f"Initiating Database ...", log="info")
self.set(key="depthcaches", value={})
self.set(key="nodes", value={})
self.set(key="pods", value={})
self.set(key="credentials", value={})
self.set(key="timestamp", value=float())
if self.app.info["name"] == "ubdcc-mgmt":
self.update_nodes()
return True
def _set_update_timestamp(self) -> bool:
self.data["timestamp"] = self.app.get_unix_timestamp()
return True
[docs]
def is_empty(self) -> bool:
if len(self.data["pods"]) == 0 and len(self.data["depthcaches"]) == 0:
return True
return False
[docs]
def add_depthcache(
self,
exchange: str = None,
market: str = None,
desired_quantity: int = None,
update_interval: int = None,
refresh_interval: int = None,
) -> bool:
if exchange is None or market is None:
raise ValueError("Missing mandatory parameter: exchange, market")
if desired_quantity is None or desired_quantity == "None":
desired_quantity = 1
else:
desired_quantity = int(desired_quantity)
if update_interval is None or update_interval == "None":
update_interval = None
else:
update_interval = int(update_interval)
if refresh_interval is None or refresh_interval == "None":
refresh_interval = None
else:
refresh_interval = int(refresh_interval)
depthcache = {
"CREATED_TIME": self.app.get_unix_timestamp(),
"DESIRED_QUANTITY": desired_quantity,
"DISTRIBUTION": {},
"DISTRIBUTION_CHANGES": 0,
"EXCHANGE": exchange,
"LAST_DISTRIBUTION_CHANGE": 0,
"REFRESH_INTERVAL": refresh_interval,
"MARKET": market,
"UPDATE_INTERVAL": update_interval,
}
with self.data_lock:
if self.data["depthcaches"].get(exchange) is None:
self.data["depthcaches"][exchange] = {}
self.data["depthcaches"][exchange][market] = depthcache
self._set_update_timestamp()
return True
[docs]
def add_depthcache_distribution(
self,
exchange: str = None,
market: str = None,
pod_uid: str = None,
scheduled_start_time: float = None,
) -> bool:
if (
exchange is None
or market is None
or pod_uid is None
or scheduled_start_time is None
):
raise ValueError(
"Missing mandatory parameter: exchange, pod_uid, market, scheduled_start_time"
)
distribution = {
"CREATED_TIME": self.app.get_unix_timestamp(),
"LAST_RESTART_TIME": 0,
"POD_UID": pod_uid,
"RESTARTS": 0,
"SCHEDULED_START_TIME": scheduled_start_time,
"STATUS": "starting",
}
with self.data_lock:
self.data["depthcaches"][exchange][market]["DISTRIBUTION"][
pod_uid
] = distribution
self.data["depthcaches"][exchange][market]["DISTRIBUTION_CHANGES"] = (
self.data["depthcaches"][exchange][market].get(
"DISTRIBUTION_CHANGES", 0
)
+ 1
)
self.data["depthcaches"][exchange][market][
"LAST_DISTRIBUTION_CHANGE"
] = self.app.get_unix_timestamp()
self._set_update_timestamp()
return True
[docs]
def add_pod(
self,
name: str = None,
uid: str = None,
node: str = None,
role: str = None,
ip: str = None,
api_port_rest: int = None,
status: str = None,
ubldc_version: str = None,
version: str = None,
) -> bool:
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
pod = {
"NAME": name,
"UID": uid,
"NODE": node,
"ROLE": role,
"IP": ip,
"API_PORT_REST": api_port_rest,
"LAST_SEEN": self.app.get_unix_timestamp(),
"STATUS": status,
"UBLDC_VERSION": ubldc_version,
"VERSION": version,
}
if ubldc_version is None:
del pod["UBLDC_VERSION"]
with self.data_lock:
self.data["pods"][uid] = pod
self._set_update_timestamp()
return True
[docs]
def add_credentials(
self, account_group: str = None, api_key: str = None, api_secret: str = None
) -> str:
"""Store a Binance API key pair for the given account group. Returns
the generated credential id (UUID). Raises ValueError on invalid
parameters or unknown account_group."""
if account_group is None or api_key is None or api_secret is None:
raise ValueError(
"Missing mandatory parameter: account_group, api_key, api_secret"
)
from .AccountGroups import is_valid_account_group
if not is_valid_account_group(account_group):
raise ValueError(f"Invalid account_group: {account_group}")
credential_id = str(uuid.uuid4())
credential = {
"ID": credential_id,
"ACCOUNT_GROUP": account_group,
"API_KEY": api_key,
"API_SECRET": api_secret,
"ADDED_TIMESTAMP": self.app.get_unix_timestamp(),
"ASSIGNED_DCNS": [],
}
with self.data_lock:
if self.data.get("credentials") is None:
self.data["credentials"] = {}
self.data["credentials"][credential_id] = credential
self._set_update_timestamp()
self.rebalance_account_group(account_group)
return credential_id
[docs]
def rebalance_account_group(self, account_group: str = None) -> bool:
"""Redistribute all active DCN uids evenly (round-robin) over the
credentials available for this account_group. Called whenever the
credential set or the DCN population for this group changes. Returns
True when the assignment actually changed."""
if account_group is None:
raise ValueError("Missing mandatory parameter: account_group")
with self.data_lock:
credentials = self.data.get("credentials") or {}
candidates = [
c for c in credentials.values() if c["ACCOUNT_GROUP"] == account_group
]
dcn_uids = sorted(
uid
for uid, pod in self.data.get("pods", {}).items()
if pod.get("ROLE") == "ubdcc-dcn"
)
before = {c["ID"]: list(c["ASSIGNED_DCNS"]) for c in candidates}
if not candidates:
# no keys for this group — nothing to do (the credentials dict
# is empty for this group anyway)
return False
for c in candidates:
c["ASSIGNED_DCNS"] = []
for i, uid in enumerate(dcn_uids):
candidates[i % len(candidates)]["ASSIGNED_DCNS"].append(uid)
after = {c["ID"]: list(c["ASSIGNED_DCNS"]) for c in candidates}
if before != after:
self._set_update_timestamp()
self.app.stdout_msg(
f"Rebalanced credentials for '{account_group}': "
f"{len(dcn_uids)} DCN(s) over {len(candidates)} key(s)",
log="info",
)
return True
return False
[docs]
def assign_credentials(
self, uid: str = None, account_group: str = None
) -> dict | None:
"""Assign (or return already-assigned) credential to a DCN uid.
Load-balanced: picks credential with fewest ASSIGNED_DCNS.
Returns full credential dict (incl. secret) or None if no keys available.
"""
if uid is None or account_group is None:
raise ValueError("Missing mandatory parameter: uid, account_group")
with self.data_lock:
credentials = self.data.get("credentials") or {}
candidates = [
c for c in credentials.values() if c["ACCOUNT_GROUP"] == account_group
]
if not candidates:
return None
for c in candidates:
if uid in c["ASSIGNED_DCNS"]:
return dict(c)
chosen = min(candidates, key=lambda c: len(c["ASSIGNED_DCNS"]))
chosen["ASSIGNED_DCNS"].append(uid)
self._set_update_timestamp()
return dict(chosen)
[docs]
def delete_credentials(self, credential_id: str = None) -> bool:
"""Remove a stored credential. Returns True on success, False when the
id was not found."""
if credential_id is None:
raise ValueError("Missing mandatory parameter: credential_id")
with self.data_lock:
credentials = self.data.get("credentials") or {}
if credential_id not in credentials:
return False
account_group = credentials[credential_id]["ACCOUNT_GROUP"]
del credentials[credential_id]
self._set_update_timestamp()
self.rebalance_account_group(account_group)
return True
[docs]
def get_credentials(self, credential_id: str = None) -> dict | None:
"""Return the full credential record (incl. secret) for the given id,
or None if not found. Internal use only — do not leak through public
endpoints."""
with self.data_lock:
credentials = self.data.get("credentials") or {}
c = credentials.get(credential_id)
return dict(c) if c else None
[docs]
def get_credentials_list(self, reveal_secrets: bool = False) -> list:
"""Return list of credentials. With reveal_secrets=False (default)
api_key/api_secret are masked for public consumption."""
result = []
with self.data_lock:
credentials = self.data.get("credentials") or {}
for c in credentials.values():
entry = {
"id": c["ID"],
"account_group": c["ACCOUNT_GROUP"],
"added_timestamp": c["ADDED_TIMESTAMP"],
"assigned_dcns": list(c["ASSIGNED_DCNS"]),
}
if reveal_secrets:
entry["api_key"] = c["API_KEY"]
entry["api_secret"] = c["API_SECRET"]
else:
entry["api_key_preview"] = self._mask(c["API_KEY"])
result.append(entry)
return result
[docs]
def release_credentials(self, uid: str = None) -> bool:
"""Remove uid from ASSIGNED_DCNS of all credentials (called when DCN
pod disappears)."""
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
with self.data_lock:
credentials = self.data.get("credentials") or {}
changed = False
for c in credentials.values():
if uid in c["ASSIGNED_DCNS"]:
c["ASSIGNED_DCNS"].remove(uid)
changed = True
if changed:
self._set_update_timestamp()
return changed
@staticmethod
def _mask(secret: str) -> str:
if not secret:
return ""
if len(secret) <= 6:
return "*" * len(secret)
return secret[:4] + "*" * (len(secret) - 6) + secret[-2:]
[docs]
def delete(self, key: str = None) -> bool:
with self.data_lock:
if key in self.data:
del self.data[key]
self._set_update_timestamp()
self.app.stdout_msg(
f"DB entry deleted: {key}", log="debug", stdout=False
)
return True
self.app.stdout_msg(f"DB entry {key} not found.", log="debug", stdout=False)
return False
[docs]
def delete_depthcache(self, exchange: str = None, market: str = None) -> bool:
if exchange is None or market is None:
raise ValueError("Missing mandatory parameter: exchange, market")
with self.data_lock:
try:
del self.data["depthcaches"][exchange][market]
except KeyError:
return True
self._set_update_timestamp()
self.app.stdout_msg(
f"DB depthcaches deleted: {exchange}, {market}", log="debug"
)
return True
[docs]
def delete_depthcache_distribution(
self, exchange: str = None, market: str = None, pod_uid: str = None
) -> bool:
if exchange is None or market is None or pod_uid is None:
raise ValueError("Missing mandatory parameter: exchange, pod_uid, market")
with self.data_lock:
del self.data["depthcaches"][exchange][market]["DISTRIBUTION"][pod_uid]
self.data["depthcaches"][exchange][market]["DISTRIBUTION_CHANGES"] = (
self.data["depthcaches"][exchange][market].get(
"DISTRIBUTION_CHANGES", 0
)
+ 1
)
self.data["depthcaches"][exchange][market][
"LAST_DISTRIBUTION_CHANGE"
] = self.app.get_unix_timestamp()
self._set_update_timestamp()
self.app.stdout_msg(
f"DB depthcache distribution deleted: {exchange}, {market}, {pod_uid}",
log="debug",
)
return True
[docs]
def delete_pod(self, uid: str = None) -> bool:
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
with self.data_lock:
del self.data["pods"][uid]
self._set_update_timestamp()
self.app.stdout_msg(f"DB pod deleted: {uid}", log="debug", stdout=True)
return True
[docs]
def delete_old_pods(self) -> bool:
old_pods = []
max_age = 60
with self.data_lock:
for uid in self.data["pods"]:
if (time.time() - max_age) > self.data["pods"][uid]["LAST_SEEN"]:
old_pods.append(uid)
for uid in old_pods:
self.delete_pod(uid=uid)
return True
[docs]
def exists_depthcache(self, exchange: str = None, market: str = None) -> bool:
if exchange is None or market is None:
raise ValueError("Missing mandatory parameter: exchange, market")
with self.data_lock:
try:
return market in self.data["depthcaches"][exchange]
except KeyError:
return False
[docs]
def exists_pod(self, uid: str = None) -> bool:
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
return uid in self.data["pods"]
[docs]
def get(self, key: str = None):
with self.data_lock:
return self.data.get(key)
[docs]
def get_all(self) -> dict:
with self.data_lock:
return self.data
[docs]
def get_available_dcn_pods(self) -> dict:
available_dcn_pods = {}
for uid in self.data["pods"]:
if self.data["pods"][uid]["ROLE"] == "ubdcc-dcn":
try:
available_dcn_pods[uid] = self.data["nodes"][
self.data["pods"][uid]["NODE"]
]["USAGE_CPU_PERCENT"]
except KeyError:
available_dcn_pods[uid] = 0
return available_dcn_pods
[docs]
def get_backup_dict(self) -> dict:
with self.data_lock:
return self.app.sort_dict(input_dict=self.app.data["db"].data)
[docs]
def get_best_dcn(
self, available_pods: dict = None, excluded_pods: list = None
) -> str | None:
if available_pods is None:
available_pods = self.get_available_dcn_pods()
for uid in excluded_pods:
try:
del available_pods[uid]
except KeyError:
pass
available_pods_uid: list = [uid for uid in available_pods.keys()]
return self.app.get_dcn_uid_unused_longest_time(selection=available_pods_uid)
[docs]
def get_dcn_responsibilities(self) -> list:
with self.data_lock:
responsibilities = []
for exchange in self.data["depthcaches"]:
for market in self.data["depthcaches"][exchange]:
for pod_uid in self.data["depthcaches"][exchange][market][
"DISTRIBUTION"
]:
if (
pod_uid == self.app.id["uid"]
and self.data["depthcaches"][exchange][market][
"DISTRIBUTION"
][pod_uid]["SCHEDULED_START_TIME"]
< self.app.get_unix_timestamp()
):
responsibilities.append(
{
"exchange": exchange,
"market": market,
"refresh_interval": self.data["depthcaches"][
exchange
][market]["REFRESH_INTERVAL"],
"update_interval": self.data["depthcaches"][
exchange
][market]["UPDATE_INTERVAL"],
}
)
return responsibilities
[docs]
def get_depthcache_list(self) -> dict:
with self.data_lock:
try:
return self.data["depthcaches"]
except KeyError:
return {}
[docs]
def get_depthcache_info(self, exchange: str = None, market: str = None) -> dict:
if exchange is None or market is None:
raise ValueError("Missing mandatory parameter: exchange, market")
with self.data_lock:
try:
return self.data["depthcaches"][exchange][market]
except KeyError:
return {}
[docs]
def get_pod_by_address(self, address: str = None) -> dict | None:
if address is None:
raise ValueError("Missing mandatory parameter: address")
with self.data_lock:
try:
for uid in self.data["pods"]:
if self.data["pods"][uid]["IP"] == address:
return self.data["pods"][uid]
except KeyError:
return None
[docs]
def get_pod_by_uid(self, uid=None) -> dict | None:
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
with self.data_lock:
try:
return self.data["pods"][uid]
except KeyError:
return None
[docs]
def get_responsible_dcn_addresses(
self, exchange: str = None, market: str = None
) -> list:
with self.data_lock:
responsible_dcn = []
try:
for pod_uid in self.data["depthcaches"][exchange][market][
"DISTRIBUTION"
]:
if (
self.data["depthcaches"][exchange][market]["DISTRIBUTION"][
pod_uid
]["STATUS"]
== "running"
):
responsible_dcn.append(
[
self.data["pods"][pod_uid]["IP"],
self.data["pods"][pod_uid]["API_PORT_REST"],
pod_uid,
]
)
except KeyError:
pass
return responsible_dcn
[docs]
def get_worst_dcn(
self, available_pods: dict | None = None, excluded_pods: list = None
) -> str | None:
if available_pods is None:
available_pods = self.get_available_dcn_pods()
# Todo: Find worst pod (cpu or subscriptions or ...)
worst_pod = None
for uid in available_pods:
if uid not in excluded_pods:
worst_pod = uid
return worst_pod
[docs]
def replace_data(self, data: dict = None):
with self.data_lock:
self.data = data
return True
[docs]
def remove_orphaned_distribution_entries(self) -> bool:
with self.data_lock:
remove_distributions = []
for exchange in self.data["depthcaches"]:
for market in self.data["depthcaches"][exchange]:
for pod_uid in self.data["depthcaches"][exchange][market][
"DISTRIBUTION"
]:
if self.exists_pod(uid=pod_uid) is False:
remove_distributions.append(
{
"exchange": exchange,
"market": market,
"pod_uid": pod_uid,
}
)
with self.data_lock:
for item in remove_distributions:
del self.data["depthcaches"][item["exchange"]][item["market"]][
"DISTRIBUTION"
][item["pod_uid"]]
self._set_update_timestamp()
return True
[docs]
def remove_orphaned_credential_assignments(self) -> bool:
with self.data_lock:
existing_pods = set(self.data.get("pods", {}).keys())
credentials = self.data.get("credentials") or {}
changed = False
for c in credentials.values():
new_list = [uid for uid in c["ASSIGNED_DCNS"] if uid in existing_pods]
if len(new_list) != len(c["ASSIGNED_DCNS"]):
c["ASSIGNED_DCNS"] = new_list
changed = True
if changed:
self._set_update_timestamp()
return True
[docs]
def revise(self) -> bool:
start_time = time.time()
self.app.stdout_msg(f"Revise the Database ...", log="info")
self.update_nodes()
self.delete_old_pods()
self.remove_orphaned_distribution_entries()
self.remove_orphaned_credential_assignments()
self.rebalance_credential_assignments_if_needed()
self.manage_distribution()
run_time = time.time() - start_time
self.app.stdout_msg(f"Database revised in {run_time} seconds!", log="info")
return True
[docs]
def rebalance_credential_assignments_if_needed(self) -> bool:
"""Detect DCN-population changes since the last revise() and rebalance
affected account_groups. Keeps every active DCN exactly one assigned
credential per account_group (or none if no key is configured)."""
with self.data_lock:
credentials = self.data.get("credentials") or {}
current_dcns = {
uid
for uid, pod in self.data.get("pods", {}).items()
if pod.get("ROLE") == "ubdcc-dcn"
}
groups_with_creds = {c["ACCOUNT_GROUP"] for c in credentials.values()}
assigned_by_group: dict[str, set[str]] = {
g: set() for g in groups_with_creds
}
for c in credentials.values():
assigned_by_group[c["ACCOUNT_GROUP"]].update(c["ASSIGNED_DCNS"])
changed = False
for group in groups_with_creds:
if assigned_by_group[group] != current_dcns:
if self.rebalance_account_group(account_group=group):
changed = True
return changed
[docs]
def manage_distribution(self) -> bool:
add_distributions = []
remove_distributions = []
with self.data_lock:
for exchange in self.data["depthcaches"]:
print(f"exchange={exchange}")
for market in self.data["depthcaches"][exchange]:
print(f"market={market}")
existing_distribution = []
for pod_uid in self.data["depthcaches"][exchange][market][
"DISTRIBUTION"
]:
existing_distribution.append(pod_uid)
existing_quantity = len(
self.data["depthcaches"][exchange][market]["DISTRIBUTION"]
)
desired_quantity = self.data["depthcaches"][exchange][market][
"DESIRED_QUANTITY"
]
if existing_quantity < desired_quantity:
add_quantity = desired_quantity - existing_quantity
delayed_start_time: float = 300.0
for i in range(0, add_quantity):
best_dcn = self.get_best_dcn(
excluded_pods=existing_distribution
)
print(f"best_dcn={best_dcn}")
if best_dcn is not None:
if existing_quantity > 0:
multiplikator = i + existing_quantity
else:
multiplikator = i
scheduled_start_time = (
self.app.get_unix_timestamp()
+ float(multiplikator) * delayed_start_time
)
add_distributions.append(
{
"exchange": exchange,
"market": market,
"pod_uid": best_dcn,
"scheduled_start_time": scheduled_start_time,
}
)
existing_distribution.append(best_dcn)
print(f"ADDED: {add_distributions}")
break
elif existing_quantity > desired_quantity:
remove_quantity = existing_quantity - desired_quantity
exclude_dcn = []
available_pods = {}
for uid in existing_distribution:
available_pods[uid] = uid
for _ in range(0, remove_quantity):
worst_dcn = self.get_worst_dcn(
available_pods=available_pods, excluded_pods=exclude_dcn
)
if worst_dcn is not None:
exclude_dcn.append(worst_dcn)
remove_distributions.append(
{
"exchange": exchange,
"market": market,
"pod_uid": worst_dcn,
}
)
for item in add_distributions:
self.add_depthcache_distribution(
exchange=item["exchange"],
market=item["market"],
pod_uid=item["pod_uid"],
scheduled_start_time=item["scheduled_start_time"],
)
for item in remove_distributions:
self.delete_depthcache_distribution(
exchange=item["exchange"],
market=item["market"],
pod_uid=item["pod_uid"],
)
return True
[docs]
def set(
self, key: str = None, value: dict | str | float | list | set | tuple = None
) -> bool:
with self.data_lock:
self.data[key] = value
self._set_update_timestamp()
self.app.stdout_msg(
f"DB entry added/updated: {key} = {value}", log="debug", stdout=False
)
return True
[docs]
def update_nodes(self) -> bool:
nodes = self.app.get_k8s_nodes()
if nodes:
self.set(key="nodes", value=nodes)
self.app.stdout_msg(f"DB all nodes updated!", log="debug", stdout=False)
return True
else:
self.app.stdout_msg(
f"Timed update of the DB key 'nodes': Query of the k8s nodes was empty, no "
f"update is performed!",
log="error",
stdout=True,
)
return False
[docs]
def update_depthcache(
self,
desired_quantity: int = None,
exchange: str = None,
refresh_interval: int = None,
market: str = None,
update_interval: int = None,
) -> bool:
if exchange is None or market is None:
raise ValueError("Missing mandatory parameter: exchange, market")
with self.data_lock:
if desired_quantity is not None:
self.data["depthcaches"][exchange][market][
"DESIRED_QUANTITY"
] = desired_quantity
self._set_update_timestamp()
if update_interval is not None:
self.data["depthcaches"][exchange][market][
"UPDATE_INTERVAL"
] = update_interval
self._set_update_timestamp()
if refresh_interval is not None:
self.data["depthcaches"][exchange][market][
"REFRESH_INTERVAL"
] = refresh_interval
self._set_update_timestamp()
self.app.stdout_msg(
f"DB depthcaches updated: {exchange}, {market}, {desired_quantity}, {update_interval}",
log="debug",
)
return True
[docs]
def update_depthcache_distribution(
self,
exchange: str = None,
market: str = None,
pod_uid: str = None,
last_restart_time: float = None,
status: str = None,
) -> bool:
if exchange is None or market is None or pod_uid is None:
raise ValueError("Missing mandatory parameter: exchange, pod_uid, market")
with self.data_lock:
if last_restart_time is not None:
try:
dist = self.data["depthcaches"][exchange][market]["DISTRIBUTION"][
pod_uid
]
except KeyError:
return False
if last_restart_time != dist.get("LAST_RESTART_TIME"):
dist["LAST_RESTART_TIME"] = last_restart_time
dist["RESTARTS"] = dist.get("RESTARTS", 0) + 1
self._set_update_timestamp()
if status is not None:
try:
self.data["depthcaches"][exchange][market]["DISTRIBUTION"][pod_uid][
"STATUS"
] = status
except KeyError:
return False
self._set_update_timestamp()
self.app.stdout_msg(
f"DB depthcache distribution updated: {exchange}, {market}, {pod_uid}, {last_restart_time},"
f" {status}",
log="debug",
)
return True
[docs]
def update_pod(
self,
uid: str = None,
node: str = None,
ip: str = None,
api_port_rest: int = None,
status: str = None,
) -> bool:
if uid is None:
raise ValueError("Missing mandatory parameter: uid")
with self.data_lock:
self.data["pods"][uid]["LAST_SEEN"] = self.app.get_unix_timestamp()
if api_port_rest is not None:
self.data["pods"][uid]["API_PORT_REST"] = api_port_rest
self._set_update_timestamp()
if ip is not None:
self.data["pods"][uid]["IP"] = ip
self._set_update_timestamp()
if node is not None:
self.data["pods"][uid]["NODE"] = node
self._set_update_timestamp()
if status is not None:
self.data["pods"][uid]["STATUS"] = status
self._set_update_timestamp()
self.app.stdout_msg(f"DB pod updated: {uid}", log="debug", stdout=False)
return True
