Make pylint happy

api/gitapi.py

@@ -1,18 +1,63 @@
-from flask import Flask, jsonify
+#!/usr/bin/env python3
+"""
+This module provides a Flask-based API for managing Git repositories in the OpenGnsys system.
+It includes endpoints for creating, deleting, synchronizing, backing up, and performing garbage
+collection on Git repositories. The API also provides endpoints for retrieving repository
+information such as the list of repositories and branches, as well as checking the status of
+asynchronous tasks.
+
+Classes:
+    None
+
+Functions:
+    do_repo_backup(repo, params)
+
+    do_repo_sync(repo, params)
+
+    do_repo_gc(repo)
+
+    home()
+
+    get_repositories()
+
+    create_repo(repo)
+
+    sync_repo(repo)
+
+    backup_repository(repo)
+
+    gc_repo(repo)
+
+    tasks_status(task_id)
+
+    delete_repo(repo)
+
+    get_repository_branches(repo)
+
+    health_check()
+
+Constants:
+    REPOSITORIES_BASE_PATH (str): The base path where Git repositories are stored.
+
+Global Variables:
+    app (Flask): The Flask application instance.
+    executor (Executor): The Flask-Executor instance for managing asynchronous tasks.
+    tasks (dict): A dictionary to store the status of asynchronous tasks.
+"""
+
+# pylint: disable=locally-disabled, line-too-long
+
 import os.path
 import os
-import git
 import shutil
-import subprocess
 import uuid
+import git
 from opengnsys_git_installer import OpengnsysGitInstaller
-from flask import Flask, request
+from flask import Flask, request, jsonify # stream_with_context, Response,
 from flask_executor import Executor
-import subprocess
-from flask import stream_with_context, Response
 import paramiko
 
-repositories_base_path = "/opt/opengnsys/images"
+REPOSITORIES_BASE_PATH = "/opt/opengnsys/images"
 
 # Create an instance of the Flask class
 app = Flask(__name__)
@@ -25,8 +70,22 @@ tasks = {}
 
 
 def do_repo_backup(repo, params):
+    """
+    Creates a backup of the specified Git repository and uploads it to a remote server via SFTP.
 
-    gitrepo = git.Repo(f"{repositories_base_path}/{repo}.git")
+    Args:
+        repo (str): The name of the repository to back up.
+        params (dict): A dictionary containing the following keys:
+            - ssh_server (str): The SSH server address.
+            - ssh_port (int): The SSH server port.
+            - ssh_user (str): The SSH username.
+            - filename (str): The remote filename where the backup will be stored.
+
+    Returns:
+        bool: True if the backup was successful.
+    """
+
+    gitrepo = git.Repo(f"{REPOSITORIES_BASE_PATH}/{repo}.git")
 
     ssh = paramiko.SSHClient()
     ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
@@ -42,24 +101,46 @@ def do_repo_backup(repo, params):
     return True
 
 def do_repo_sync(repo, params):
-    gitrepo = git.Repo(f"{repositories_base_path}/{repo}.git")
+    """
+    Synchronizes a local Git repository with a remote repository.
+
+    Args:
+        repo (str): The name of the local repository to synchronize.
+        params (dict): A dictionary containing the remote repository URL with the key "remote_repository".
+
+    Returns:
+        list: A list of dictionaries, each containing:
+            - "local_ref" (str): The name of the local reference.
+            - "remote_ref" (str): The name of the remote reference.
+            - "summary" (str): A summary of the push operation for the reference.
+    """
+    gitrepo = git.Repo(f"{REPOSITORIES_BASE_PATH}/{repo}.git")
 
     # Recreate the remote every time, it might change
     if "backup" in gitrepo.remotes:
         gitrepo.delete_remote("backup")
 
     backup_repo = gitrepo.create_remote("backup", params["remote_repository"])
-    pushrets = backup_repo.push("*:*")
+    pushed_references = backup_repo.push("*:*")
     results = []
 
     # This gets returned to the API
-    for ret in pushrets:
+    for ref in pushed_references:
-        results = results + [ {"local_ref" : ret.local_ref.name, "remote_ref" : ret.remote_ref.name, "summary" : ret.summary }]
+        results = results + [ {"local_ref" : ref.local_ref.name, "remote_ref" : ref.remote_ref.name, "summary" : ref.summary }]
 
     return results
 
 def do_repo_gc(repo):
-    gitrepo = git.Repo(f"{repositories_base_path}/{repo}.git")
+    """
+    Perform garbage collection on the specified Git repository.
+
+    Args:
+        repo (str): The name of the repository to perform garbage collection on.
+
+    Returns:
+        bool: True if the garbage collection command was executed successfully.
+    """
+    gitrepo = git.Repo(f"{REPOSITORIES_BASE_PATH}/{repo}.git")
 
     gitrepo.git.gc()
     return True
@@ -99,11 +180,11 @@ def get_repositories():
     }
     """
 
-    if not os.path.isdir(repositories_base_path):
+    if not os.path.isdir(REPOSITORIES_BASE_PATH):
         return jsonify({"error": "Repository storage not found, git functionality may not be installed."}), 500
 
     repos = []
-    for entry in os.scandir(repositories_base_path):
+    for entry in os.scandir(REPOSITORIES_BASE_PATH):
         if entry.is_dir(follow_symlinks=False) and os.path.isfile(os.path.join(entry.path, "HEAD")):
             name = entry.name
             if name.endswith(".git"):
@@ -131,13 +212,13 @@ def create_repo(repo):
             - 200: If the repository already exists.
             - 201: If the repository is successfully created.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if os.path.isdir(repo_path):
         return jsonify({"status": "Repository already exists"}), 200
 
 
     installer = OpengnsysGitInstaller()
-    installer._init_git_repo(repo + ".git")
+    installer.init_git_repo(repo + ".git")
 
 
     return jsonify({"status": "Repository created"}), 201
@@ -160,7 +241,7 @@ def sync_repo(repo):
         - 400: If the request payload is missing or invalid.
         - 404: If the specified repository is not found.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if not os.path.isdir(repo_path):
         return jsonify({"error": "Repository not found"}), 404
 
@@ -170,9 +251,6 @@ def sync_repo(repo):
     if data is None:
         return jsonify({"error" : "Parameters missing"}), 400
 
-    dest_repo = data["remote_repository"]
-
-
     future = executor.submit(do_repo_sync, repo, data)
     task_id = str(uuid.uuid4())
     tasks[task_id] = future
@@ -180,7 +258,7 @@ def sync_repo(repo):
 
 
 @app.route('/repositories/<repo>/backup', methods=['POST'])
-def backup_repo(repo):
+def backup_repository(repo):
     """
     Backup a specified repository.
 
@@ -203,7 +281,7 @@ def backup_repo(repo):
         - The backup operation is performed asynchronously using a thread pool executor.
         - The task ID of the backup operation is generated using UUID and stored in a global tasks dictionary.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if not os.path.isdir(repo_path):
         return jsonify({"error": "Repository not found"}), 404
 
@@ -240,7 +318,7 @@ def gc_repo(repo):
                 a unique task ID if the repository is found, or an error
                 message if the repository is not found.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if not os.path.isdir(repo_path):
         return jsonify({"error": "Repository not found"}), 404
 
@@ -295,7 +373,7 @@ def delete_repo(repo):
     Returns:
         Response: A JSON response with a status message and the appropriate HTTP status code.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if not os.path.isdir(repo_path):
         return jsonify({"error": "Repository not found"}), 404
 
@@ -319,14 +397,14 @@ def get_repository_branches(repo):
         - 200: A JSON object with a "branches" key containing a list of branch names.
         - 404: A JSON object with an "error" key containing the message "Repository not found" if the repository does not exist.
     """
-    repo_path = os.path.join(repositories_base_path, repo + ".git")
+    repo_path = os.path.join(REPOSITORIES_BASE_PATH, repo + ".git")
     if not os.path.isdir(repo_path):
         return jsonify({"error": "Repository not found"}), 404
 
-    gitRepo = git.Repo(repo_path)
+    git_repo = git.Repo(repo_path)
 
     branches = []
-    for branch in gitRepo.branches:
+    for branch in git_repo.branches:
         branches = branches + [branch.name]
 
 
@@ -357,5 +435,3 @@ def health_check():
 # Run the Flask app
 if __name__ == '__main__':
     app.run(debug=True, host='0.0.0.0')
-
-

installer/opengnsys_git_installer.py

@@ -157,7 +157,7 @@ class OpengnsysGitInstaller:
         if self.temp_dir:
             shutil.rmtree(self.temp_dir, ignore_errors=True)
 
-    def _init_git_repo(self, reponame):
+    def init_git_repo(self, reponame):
         """Inicializa un repositorio Git"""
         # Creamos repositorio
         ogdir_images = os.path.join(self.base_path, "images")
@@ -330,9 +330,9 @@ class OpengnsysGitInstaller:
         os.system(f"usermod -s {SHELL} opengnsys")
 
         # Creamos repositorios
-        self._init_git_repo('windows.git')
+        self.init_git_repo('windows.git')
-        self._init_git_repo('linux.git')
+        self.init_git_repo('linux.git')
-        self._init_git_repo('mac.git')
+        self.init_git_repo('mac.git')
 
         # Damos permiso al usuario opengnsys
         for DIR in ["base.git", "linux.git", "windows.git"]: #, "LinAcl", "WinAcl"]: