Source code for cobbler.cobbler_collections.manager

Repository of the Cobbler object model

# 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 weakref
from typing import TYPE_CHECKING, Any, Dict, Union

from cobbler import serializer, validate
from cobbler.cexceptions import CX
from cobbler.cobbler_collections.distros import Distros
from cobbler.cobbler_collections.images import Images
from cobbler.cobbler_collections.menus import Menus
from cobbler.cobbler_collections.profiles import Profiles
from cobbler.cobbler_collections.repos import Repos
from import Systems
from cobbler.items.item import Item
from cobbler.settings import Settings

    from cobbler.api import CobblerAPI
    from cobbler.cobbler_collections.collection import ITEM, Collection

    COLLECTION_UNION = Union[Menus, Distros, Repos, Profiles, Images, Systems]

[docs]class CollectionManager: """ Manages a definitive copy of all data cobbler_collections with weakrefs pointing back into the class so they can understand each other's contents. """ has_loaded = False __shared_state: Dict[str, Any] = {} def __init__(self, api: "CobblerAPI") -> None: """ Constructor which loads all content if this action was not performed before. """ self.__dict__ = CollectionManager.__shared_state if not CollectionManager.has_loaded: self.__load(api) def __load(self, api: "CobblerAPI") -> None: """ Load all collections from the disk into Cobbler. :param api: The api to resolve information with. """ CollectionManager.has_loaded = True self.api = api self.__serializer = serializer.Serializer(api) self._distros = Distros(weakref.proxy(self)) self._repos = Repos(weakref.proxy(self)) self._profiles = Profiles(weakref.proxy(self)) self._systems = Systems(weakref.proxy(self)) self._images = Images(weakref.proxy(self)) self._menus = Menus(weakref.proxy(self))
[docs] def distros(self) -> Distros: """ Return the definitive copy of the Distros collection """ return self._distros
[docs] def profiles(self) -> Profiles: """ Return the definitive copy of the Profiles collection """ return self._profiles
[docs] def systems(self) -> Systems: """ Return the definitive copy of the Systems collection """ return self._systems
[docs] def settings(self) -> "Settings": """ Return the definitive copy of the application settings """ return self.api.settings()
[docs] def repos(self) -> Repos: """ Return the definitive copy of the Repos collection """ return self._repos
[docs] def images(self) -> Images: """ Return the definitive copy of the Images collection """ return self._images
[docs] def menus(self) -> Menus: """ Return the definitive copy of the Menus collection """ return self._menus
[docs] def serialize(self) -> None: """ Save all cobbler_collections to disk """ self.__serializer.serialize(self._distros) self.__serializer.serialize(self._repos) self.__serializer.serialize(self._profiles) self.__serializer.serialize(self._images) self.__serializer.serialize(self._systems) self.__serializer.serialize(self._menus)
[docs] def serialize_one_item(self, item: "ITEM") -> None: # type: ignore """ Save a collection item to disk :param item: collection item """ collection: "Collection[ITEM]" = self.get_items(item.COLLECTION_TYPE) # type: ignore self.__serializer.serialize_item(collection, item)
[docs] def serialize_item(self, collection: "Collection[ITEM]", item: "ITEM") -> None: """ Save a collection item to disk Deprecated - Use above serialize_one_item function instead collection param can be retrieved :param collection: Collection :param item: collection item """ self.__serializer.serialize_item(collection, item)
[docs] def serialize_delete_one_item(self, item: "ITEM") -> None: # type: ignore """ Save a collection item to disk :param item: collection item """ collection: "COLLECTION_UNION" = self.get_items(item.COLLECTION_TYPE) # type: ignore self.__serializer.serialize_delete(collection, item) # type: ignore
[docs] def serialize_delete(self, collection: "Collection[ITEM]", item: "ITEM") -> None: """ Delete a collection item from disk :param collection: collection :param item: collection item """ self.__serializer.serialize_delete(collection, item)
[docs] def deserialize(self) -> None: """ Load all cobbler_collections from disk :raises CX: if there is an error in deserialization """ collection: "COLLECTION_UNION" for args in ( (self._menus, True), (self._distros, False), (self._repos, False), (self._profiles, True), (self._images, False), (self._systems, False), ): try: self.__serializer.deserialize(collection=args[0], topological=args[1]) # type: ignore except Exception as error: raise CX( f"serializer: error loading collection {collection.collection_type()}: {error}." f"Check your settings!" ) from error
[docs] def deserialize_one_item(self, obj: Item) -> Dict[str, Any]: """ Load a collection item from disk :param obj: collection item """ collection_type = self.get_items(obj.COLLECTION_TYPE).collection_types() return self.__serializer.deserialize_item(collection_type,
[docs] def get_items( self, collection_type: str ) -> Union[Distros, Profiles, Systems, Repos, Images, Menus, "Settings"]: """ Get a full collection of a single type. Valid Values vor ``collection_type`` are: "distro", "profile", "repo", "image", "menu" and "settings". :param collection_type: The type of collection to return. :return: The collection if ``collection_type`` is valid. :raises CX: If the ``collection_type`` is invalid. """ result: Union[ Distros, Profiles, Systems, Repos, Images, Menus, "Settings", ] if validate.validate_obj_type(collection_type) and hasattr( self, f"_{collection_type}s" ): result = getattr(self, f"_{collection_type}s") else: raise CX( f'internal error, collection name "{collection_type}" not supported' ) return result