persistence
Atomic YAML persistence with per-file locks and change watching.
SealSkin keeps its configuration in hand-editable YAML files. This module is the single place that reads and writes them so every writer gets:
- atomic replacement (temporary file in the same directory +
os.replace), - one
Lockper path so concurrent handlers never interleave, - a content hash of the last write so the file watcher can tell our own writes apart from edits made by an administrator.
attributelogger= logging.getLogger(__name__)attributeReloadCallback= Callable[[str], Awaitable[None]]funclock_for(path) -> asyncio.LockReturn the lock guarding path, creating it on first use.
parampathstrFile path (normalised with abspath).
Returns
asyncio.LockThe Lock shared by every writer of that file.
funccontent_hash(data) -> strReturn the SHA-256 hex digest of data.
paramdatabytesReturns
strfuncread_yaml(path, default=None) -> AnyLoad a YAML file.
parampathstrFile to read.
paramdefaultAny= NoneValue returned when the file does not exist or is empty.
Returns
typing.AnyThe parsed document, or default.
funcdump_yaml(data) -> strSerialise data the way every SealSkin file is written.
paramdataAnyAny YAML-serialisable structure.
Returns
strYAML text with keys in insertion order.
funcwrite_yaml_sync(path, data) -> NoneAtomically write data to path as YAML.
The document is written to a temporary file in the target directory and
moved into place with replace, so readers never observe a
partially written file.
parampathstrDestination file.
paramdataAnyYAML-serialisable structure.
Returns
Nonefuncwrite_yaml(path, data) -> NoneAtomically write data to path under the file's lock.
parampathstrDestination file.
paramdataAnyYAML-serialisable structure.
Returns
Nonefuncwas_written_by_us(path) -> boolTell whether the current contents of path match our last write.
parampathstrFile to check.
Returns
boolTrue when the on-disk content hash equals the hash recorded by
funcwatch_paths(targets, stop_event, debounce_ms=500) -> NoneWatch files and directories and call a callback when they change.
paramtargetsdict[str, ReloadCallback]Mapping of path (file or directory) to the coroutine function invoked with the changed path. Directory targets fire for any file inside them.
paramstop_eventasyncio.EventSet it to end the watch loop.
paramdebounce_msint= 500Quiet period before a batch of changes is reported.
Returns
None