"""
Repository of the Cobbler object model
Copyright 2006-2009, Red Hat, Inc and Others
Michael DeHaan <michael.dehaan AT gmail>
This program is free software; you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation; either version 2 of the License, or
(at your option) any later version.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
02110-1301 USA
"""
import weakref
from typing import TYPE_CHECKING, Union, Dict, Any
from cobbler.cexceptions import CX
from cobbler import serializer
from cobbler import validate
from cobbler.settings import Settings
from cobbler.cobbler_collections.distros import Distros
from cobbler.cobbler_collections.files import Files
from cobbler.cobbler_collections.images import Images
from cobbler.cobbler_collections.mgmtclasses import Mgmtclasses
from cobbler.cobbler_collections.packages import Packages
from cobbler.cobbler_collections.profiles import Profiles
from cobbler.cobbler_collections.repos import Repos
from cobbler.cobbler_collections.systems import Systems
from cobbler.cobbler_collections.menus import Menus
if TYPE_CHECKING:
from cobbler.api import CobblerAPI
[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"):
"""
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"):
"""
Load all collections from the disk into Cobbler.
:param api: The api to resolve information with.
"""
CollectionManager.has_loaded = True
self.api = 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._mgmtclasses = Mgmtclasses(weakref.proxy(self))
self._packages = Packages(weakref.proxy(self))
self._files = Files(weakref.proxy(self))
self._menus = Menus(weakref.proxy(self))
[docs]
def distros(self):
"""
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):
"""
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 mgmtclasses(self) -> Mgmtclasses:
"""
Return the definitive copy of the Mgmtclasses collection
"""
return self._mgmtclasses
[docs]
def packages(self) -> Packages:
"""
Return the definitive copy of the Packages collection
"""
return self._packages
[docs]
def files(self) -> Files:
"""
Return the definitive copy of the Files collection
"""
return self._files
[docs]
def serialize(self):
"""
Save all cobbler_collections to disk
"""
serializer.serialize(self._distros)
serializer.serialize(self._repos)
serializer.serialize(self._profiles)
serializer.serialize(self._images)
serializer.serialize(self._systems)
serializer.serialize(self._mgmtclasses)
serializer.serialize(self._packages)
serializer.serialize(self._files)
serializer.serialize(self._menus)
# pylint: disable=R0201
[docs]
def serialize_one_item(self, item):
"""
Save a collection item to disk
:param item: collection item
"""
collection = self.get_items(item.COLLECTION_TYPE)
serializer.serialize_item(collection, item)
# pylint: disable=R0201
[docs]
def serialize_item(self, collection, item):
"""
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
"""
serializer.serialize_item(collection, item)
# pylint: disable=R0201
[docs]
def serialize_delete_one_item(self, item):
"""
Save a collection item to disk
:param item: collection item
"""
collection = self.get_items(item.COLLECTION_TYPE)
serializer.serialize_delete(collection, item)
# pylint: disable=R0201
[docs]
def serialize_delete(self, collection, item):
"""
Delete a collection item from disk
:param collection: collection
:param item: collection item
"""
serializer.serialize_delete(collection, item)
[docs]
def deserialize(self):
"""
Load all cobbler_collections from disk
:raises CX: if there is an error in deserialization
"""
old_cache_enabled = self.api.settings().cache_enabled
self.api.settings().cache_enabled = False
for args in (
(self._menus, True),
(self._distros, False),
(self._repos, False),
(self._profiles, True),
(self._images, False),
(self._systems, False),
(self._mgmtclasses, False),
(self._packages, False),
(self._files, False),
):
try:
serializer.deserialize(collection=args[0], topological=args[1])
except Exception as error:
raise CX(
f"serializer: error loading collection {args[0].collection_type()}: {error}."
f"Check your settings!"
) from error
self.api.settings().cache_enabled = old_cache_enabled
[docs]
def deserialize_one_item(self, obj) -> dict:
"""
Load a collection item from disk
:param obj: collection item
"""
collection_type = self.get_items(obj.COLLECTION_TYPE).collection_types()
return serializer.deserialize_item(collection_type, obj.name)
[docs]
def get_items(
self, collection_type: str
) -> Union[
Distros,
Profiles,
Systems,
Repos,
Images,
Mgmtclasses,
Packages,
Files,
Menus,
Settings,
]:
"""
Get a full collection of a single type.
Valid Values vor ``collection_type`` are: "distro", "profile", "repo", "image", "mgmtclass", "package", "file"
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,
Mgmtclasses,
Packages,
Files,
Menus,
Settings,
]
if validate.validate_obj_type(collection_type) and hasattr(
self, f"_{collection_type}s"
):
result = getattr(self, f"_{collection_type}s")
elif collection_type == "mgmtclass":
result = self._mgmtclasses
elif collection_type == "settings":
result = self.api.settings()
else:
raise CX(
'internal error, collection name "%s" not supported' % collection_type
)
return result