doc: ai update state_reader.py

pyaptly/state_reader.py

@@ -9,7 +9,8 @@
 
 
 class SystemStateReader(object):
-    """Reads the state from aptly and gpg.
+    """
+    Read the state from aptly and gpg.
 
     To find out what operations have to be performed to reach the state defined
     in the toml config-file.
@@ -18,6 +19,13 @@ class SystemStateReader(object):
     known_dependency_types = ("repo", "snapshot", "mirror", "gpg_key")
 
     def __init__(self):
+        """
+        Initialize the SystemStateReader.
+
+        With empty sets and dictionaries for tracking GPG keys, mirrors,
+        repositories, snapshots, and publishes, as well as mappings for
+        snapshots and publishes.
+        """
         self.gpg_keys = set()
         self.mirrors = set()
         self.repos = set()
@@ -27,9 +35,11 @@ def __init__(self):
         self.publish_map = {}
 
     def _extract_sources(self, data):
-        """Extract sources from data.
+        """
+        Extract sources from `data`.
+
+        `data` needs to be in the following format:
 
-        Data needs to be in following format:
         Name: test-snap
         Description: some description
         Sources:
@@ -51,7 +61,14 @@ def _extract_sources(self, data):
         return sources
 
     def read(self):
-        """Read all available system states."""
+        """
+        Read all available system states.
+
+        This method sequentially reads the current system states for GPG keys,
+        repositories, mirrors, snapshots, snapshot mappings, publishes, and
+        publish mappings by invoking the respective read methods for each
+        entity.
+        """
         self.read_gpg()
         self.read_repos()
         self.read_mirror()
@@ -61,7 +78,12 @@ def read(self):
         self.read_publish_map()
 
     def read_gpg(self):
-        """Read all trusted keys in gpg."""
+        """
+        Read all trusted keys in gpg.
+
+        This method populates 'self.gpg_keys' with the set of all trusted GPG
+        keys found.
+        """
         self.gpg_keys = set()
         cmd = [
             "gpg",
@@ -81,7 +103,13 @@ def read_gpg(self):
                 self.gpg_keys.add(key_short)
 
     def read_publish_map(self):
-        """Create a publish map. publish -> snapshots."""
+        """
+        Create a mapping of publishes to snapshots.
+
+        This method constructs a dictionary where each key is a publish entity,
+        and its value is a set of snapshots associated with that publish. This
+        mapping is stored in 'self.publish_map'.
+        """
         self.publish_map = {}
         # match example:  main: test-snapshot [snapshot]
         re_snap = re.compile(r"\s+[\w\d-]+\:\s([\w\d-]+)\s\[snapshot\]")
@@ -97,9 +125,12 @@ def read_publish_map(self):
         lg.debug("Joined snapshots and publishes: %s", self.publish_map)
 
     def read_snapshot_map(self):
-        """Create a snapshot map. snapshot -> snapshots.
+        """
+        Create a mapping of snapshots to their source snapshots.
 
-        This is also called merge-tree.
+        This method populates 'self.snapshot_map' with a mapping where each key
+        is a snapshot and its value is a set of snapshots that are sources of
+        the given snapshot.
         """
         self.snapshot_map = {}
         # match example:  test-snapshot [snapshot]
@@ -116,32 +147,31 @@ def read_snapshot_map(self):
         lg.debug("Joined snapshots with self(snapshots): %s", self.snapshot_map)
 
     def read_publishes(self):
-        """Read all available publishes."""
+        """Read all available publishes and store them in 'self.publishes'."""
         self.publishes = set()
         self.read_aptly_list("publish", self.publishes)
 
     def read_repos(self):
-        """Read all available repos."""
+        """Read all available repos and store them in 'self.repos'."""
         self.repos = set()
         self.read_aptly_list("repo", self.repos)
 
     def read_mirror(self):
-        """Read all available mirrors."""
+        """Read all available mirrors and store them in 'self.mirrors'."""
         self.mirrors = set()
         self.read_aptly_list("mirror", self.mirrors)
 
     def read_snapshot(self):
-        """Read all available snapshots."""
+        """Read all available snapshots and store them in 'self.snapshots'."""
         self.snapshots = set()
         self.read_aptly_list("snapshot", self.snapshots)
 
     def read_aptly_list(self, type_, list_):
-        """Read lists from aptly.
+        """
+        Read lists from aptly and populate `list_` with the results.
 
-        :param type_: The type of list to read ie. snapshot
-        :type  type_: str
-        :param list_: Read into this list
-        :param list_: list
+        The type of list to read, such as 'snapshot', is specified by `type_`.
+        The results are added to the set provided in `list_`.
         """
         cmd = ["aptly", type_, "list", "-raw"]
         result = util.run_command(cmd, stdout=util.PIPE, check=True)
@@ -151,10 +181,14 @@ def read_aptly_list(self, type_, list_):
                 list_.add(clean_line)
 
     def has_dependency(self, dependency):
-        """Check system state dependencies.
+        """
+        Check system state dependencies.
 
-        :param dependency: The dependency to check
-        :type  dependency: list
+        Determines if the specified `dependency` exists within the system's
+        current state. The `dependency` should be a list where the first element
+        is the type of dependency (e.g., 'repo', 'mirror', 'snapshot',
+        'gpg_key', 'virtual') and the second element is the name of the
+        dependency.
         """
         type_, name = dependency
 
@@ -179,6 +213,12 @@ def has_dependency(self, dependency):
 
 
 def state_reader():
+    """
+    Provides a singleton instance of `SystemStateReader`.
+
+    This function ensures that only one instance of `SystemStateReader` is
+    created and returned upon each call, implementing the singleton pattern.
+    """
     global _state_reader
     if not _state_reader:
         _state_reader = SystemStateReader()