mirror of
https://github.com/clearml/clearml-server
synced 2025-01-31 10:56:48 +00:00
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
from typing import Any, Optional, Sequence, Tuple
|
|
|
|
from mongoengine import Document, StringField, DynamicField, Q
|
|
from mongoengine.errors import NotUniqueError
|
|
|
|
from apiserver.database import Database, strict
|
|
from apiserver.database.model import DbModelMixin
|
|
|
|
|
|
class SettingKeys:
|
|
server__uuid = "server.uuid"
|
|
|
|
|
|
class Settings(DbModelMixin, Document):
|
|
meta = {
|
|
"db_alias": Database.backend,
|
|
"strict": strict,
|
|
}
|
|
|
|
key = StringField(primary_key=True)
|
|
value = DynamicField()
|
|
|
|
@classmethod
|
|
def get_by_key(cls, key: str, default: Optional[Any] = None, sep: str = ".") -> Any:
|
|
key = key.strip(sep)
|
|
res = Settings.objects(key=key).first()
|
|
if not res:
|
|
return default
|
|
return res.value
|
|
|
|
@classmethod
|
|
def get_by_prefix(
|
|
cls, key_prefix: str, default: Optional[Any] = None, sep: str = "."
|
|
) -> Sequence[Tuple[str, Any]]:
|
|
key_prefix = key_prefix.strip(sep)
|
|
query = Q(key=key_prefix) | Q(key__startswith=key_prefix + sep)
|
|
res = Settings.objects(query)
|
|
if not res:
|
|
return default
|
|
return [(x.key, x.value) for x in res]
|
|
|
|
@classmethod
|
|
def set_or_add_value(cls, key: str, value: Any, sep: str = ".") -> bool:
|
|
""" Sets a new value or adds a new key/value setting (if key does not exist) """
|
|
key = key.strip(sep)
|
|
res = Settings.objects(key=key).update(key=key, value=value, upsert=True)
|
|
return bool(res)
|
|
|
|
@classmethod
|
|
def add_value(cls, key: str, value: Any, sep: str = ".") -> bool:
|
|
""" Adds a new key/value settings. Fails if key already exists. """
|
|
key = key.strip(sep)
|
|
try:
|
|
res = cls(key=key, value=value).save(force_insert=True)
|
|
return bool(res)
|
|
except NotUniqueError:
|
|
return False
|