"""
Serializer code for Cobbler
"""
# SPDX-License-Identifier: GPL-2.0-or-later
# SPDX-FileCopyrightText: Copyright 2006-2009, Red Hat, Inc and Others
# SPDX-FileCopyrightText: Michael DeHaan <michael.dehaan AT gmail>
import fcntl
import logging
import os
import pathlib
import sys
import time
from types import ModuleType
from typing import TYPE_CHECKING, Any, Dict, Optional, TextIO
if TYPE_CHECKING:
from cobbler.api import CobblerAPI
from cobbler.cobbler_collections.collection import ITEM, Collection
[docs]class Serializer:
"""
Serializer interface that is used to access data in Cobbler independent of the actual data source.
"""
def __init__(self, api: "CobblerAPI"):
"""
Constructor that created the state for the object.
:param api: The CobblerAPI that is used for accessing shared functionality.
"""
self.api = api
self.logger = logging.getLogger()
self.lock_enabled = True
self.lock_handle: Optional[TextIO] = None
self.lock_file_location = "/var/lib/cobbler/lock"
self.storage_module = self.__get_storage_module()
self.storage_object = self.storage_module.storage_factory(api)
def __grab_lock(self) -> None:
"""
Dual purpose locking:
(A) flock to avoid multiple process access
(B) block signal handler to avoid ctrl+c while writing YAML
"""
try:
if self.lock_enabled:
if not os.path.exists(self.lock_file_location):
pathlib.Path(self.lock_file_location).touch()
self.lock_handle = open(self.lock_file_location, "r", encoding="UTF-8")
fcntl.flock(self.lock_handle.fileno(), fcntl.LOCK_EX)
except Exception as exception:
# this is pretty much FATAL, avoid corruption and quit now.
self.logger.exception("File locking error.", exc_info=exception)
sys.exit(7)
def __release_lock(self, with_changes: bool = False) -> None:
"""
Releases the lock on the resource that is currently being written.
:param with_changes: If this is true the global modification time is being updated. Default is false.
"""
if with_changes:
# this file is used to know the time of last modification on cobbler_collections
# was made -- allowing the API to work more smoothly without
# a lot of unnecessary reloads.
with open(self.api.mtime_location, "w", encoding="UTF-8") as mtime_fd:
mtime_fd.write(f"{time.time():f}")
if self.lock_enabled:
self.lock_handle = open(self.lock_file_location, "r", encoding="UTF-8")
fcntl.flock(self.lock_handle.fileno(), fcntl.LOCK_UN)
self.lock_handle.close()
[docs] def serialize(self, collection: "Collection[ITEM]") -> None:
"""
Save a collection to disk
:param collection: The collection to serialize.
"""
self.__grab_lock()
self.storage_object.serialize(collection)
self.__release_lock()
[docs] def serialize_item(self, collection: "Collection[ITEM]", item: "ITEM") -> None:
"""
Save a collection item to disk
:param collection: The Cobbler collection to know the type of the item.
:param item: The collection item to serialize.
"""
self.__grab_lock()
self.storage_object.serialize_item(collection, item)
self.__release_lock(with_changes=True)
[docs] def serialize_delete(self, collection: "Collection[ITEM]", item: "ITEM") -> None:
"""
Delete a collection item from disk
:param collection: The Cobbler collection to know the type of the item.
:param item: The collection item to delete.
"""
self.__grab_lock()
self.storage_object.serialize_delete(collection, item)
self.__release_lock(with_changes=True)
[docs] def deserialize(
self, collection: "Collection[ITEM]", topological: bool = True
) -> None:
"""
Load a collection from disk.
:param collection: The Cobbler collection to know the type of the item.
:param topological: Sort collection based on each items' depth attribute in the list of collection items. This
ensures properly ordered object loading from disk with objects having parent/child
relationships, i.e. profiles/subprofiles. See cobbler/items/item.py
"""
self.__grab_lock()
self.storage_object.deserialize(collection, topological)
self.__release_lock()
[docs] def deserialize_item(self, collection_type: str, item_name: str) -> Dict[str, Any]:
"""
Load a collection item from disk.
:param collection_type: The collection type to deserialize.
:param item_name: The collection item name to deserialize.
"""
self.__grab_lock()
result = self.storage_object.deserialize_item(collection_type, item_name)
self.__release_lock()
return result
def __get_storage_module(self) -> ModuleType:
"""
Look up configured module in the settings
:returns: A Python module.
"""
return self.api.get_module_from_file(
"serializers",
self.api.settings()
.modules.get("serializers", {})
.get("module", "serializers.file"),
"serializers.file",
)