Remove unused complex functionality from settings_manager.py.
This commit is contained in:
parent
816b6fc82b
commit
d19d44ba4f
1 changed files with 3 additions and 96 deletions
|
|
@ -7,42 +7,12 @@ the server.
|
|||
|
||||
from collections.abc import Mapping
|
||||
from socket import gethostname
|
||||
from typing import Annotated, Any, MutableMapping, Optional, Sequence
|
||||
from typing import Optional
|
||||
from uuid import UUID, uuid4
|
||||
from fastapi import Depends, HTTPException, Request
|
||||
|
||||
import labthings_fastapi as lt
|
||||
|
||||
|
||||
def thing_server_from_request(request: Request) -> lt.ThingServer:
|
||||
"""Wrap lt.find_thing_server."""
|
||||
return lt.find_thing_server(request.app)
|
||||
|
||||
|
||||
ThingServerDep = Annotated[lt.ThingServer, Depends(thing_server_from_request)]
|
||||
|
||||
|
||||
def recursive_update(old_dict: MutableMapping, update: Mapping):
|
||||
"""Update a dictionary recursively."""
|
||||
for k, v in update.items():
|
||||
if isinstance(v, Mapping):
|
||||
if k in old_dict and isinstance(old_dict[k], MutableMapping):
|
||||
recursive_update(old_dict[k], v)
|
||||
else:
|
||||
old_dict[k] = dict(**v)
|
||||
else:
|
||||
old_dict[k] = v
|
||||
|
||||
|
||||
def nested_dict_get(data: dict, key: Sequence[str], create=False) -> Any:
|
||||
"""Index a dict-of-dicts with a sequence of strings."""
|
||||
subdict = data
|
||||
for k in key:
|
||||
if k not in subdict and create:
|
||||
subdict[k] = {}
|
||||
subdict = subdict[k]
|
||||
return subdict
|
||||
|
||||
|
||||
class SettingsManager(lt.Thing):
|
||||
"""Provides functionality to other Things about the current server state.
|
||||
|
||||
|
|
@ -50,59 +20,6 @@ class SettingsManager(lt.Thing):
|
|||
and the state of other Things.
|
||||
"""
|
||||
|
||||
external_metadata = lt.ThingSetting(
|
||||
initial_value={},
|
||||
model=Mapping,
|
||||
)
|
||||
"""External metadata stored in the server's settings."""
|
||||
|
||||
external_metadata_in_state = lt.ThingSetting(
|
||||
initial_value=[],
|
||||
model=Sequence[str],
|
||||
)
|
||||
"""Keys from ``external_metadata`` that are included in the "state" metadata."""
|
||||
|
||||
@lt.thing_action
|
||||
def update_external_metadata(
|
||||
self, data: Mapping, key: Optional[str] = None
|
||||
) -> None:
|
||||
"""Add or replace keys in the external metadata.
|
||||
|
||||
The data supplied will be merged into the existing dictionary
|
||||
recursively, i.e. if a key exists, it will be added to rather than
|
||||
replaced.
|
||||
|
||||
If a key is supplied, we will treat ``data`` as being relative to that
|
||||
key, i.e. calling this action with ``data={"a":1 }, key="foo/bar"`` is
|
||||
equivalent to calling it with ``data={"foo": {"bar": {"a": 1}}}``.
|
||||
"""
|
||||
metadata = self.external_metadata
|
||||
subdict = metadata
|
||||
# If we're operating only on a part of the dict, make sure
|
||||
# that key exists, and find it.
|
||||
if key:
|
||||
subdict = nested_dict_get(subdict, key.split("/"), create=True)
|
||||
recursive_update(subdict, data)
|
||||
self.external_metadata = metadata
|
||||
|
||||
@lt.thing_action
|
||||
def delete_external_metadata(self, key: str) -> None:
|
||||
"""Delete a key from the stored metadata.
|
||||
|
||||
The key may contain forward slashes, which are understood to separate
|
||||
levels of the dictionary - i.e. ``'a/c'`` will remove the ``c`` key from a
|
||||
dictionary that looks like: ``{'a': {'c': 1}, 'b': {'d': 2}}``
|
||||
"""
|
||||
metadata = self.external_metadata
|
||||
try:
|
||||
subdict = nested_dict_get(metadata, key.split("/")[:-1])
|
||||
del subdict[key.split("/")[-1]]
|
||||
except KeyError:
|
||||
raise HTTPException(
|
||||
status_code=404, detail="The specified key '{key}' was not found"
|
||||
)
|
||||
self.external_metadata = metadata
|
||||
|
||||
_microscope_id: Optional[str] = None
|
||||
|
||||
@lt.thing_setting
|
||||
|
|
@ -130,17 +47,7 @@ class SettingsManager(lt.Thing):
|
|||
@property
|
||||
def thing_state(self) -> Mapping:
|
||||
"""Summary metadata describing the current state of the Thing."""
|
||||
state = {
|
||||
return {
|
||||
"hostname": self.hostname,
|
||||
"microscope-uuid": str(self.microscope_id),
|
||||
}
|
||||
metadata = self.external_metadata
|
||||
for k in self.external_metadata_in_state:
|
||||
try:
|
||||
kl = k.split("/")
|
||||
v = nested_dict_get(metadata, kl)
|
||||
subdict = nested_dict_get(state, kl[:-1], create=True)
|
||||
subdict[kl[-1]] = v
|
||||
except KeyError:
|
||||
continue
|
||||
return state
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue