feat: Use gitlab module vs requests to modify cicd vars

scripts/gitlab-ci-vars-reader.py

-import requests
-import json
 import argparse
+import json
+
+import gitlab
 
-# Load configuration from JSON file
-def load_config(file_path):
-    try:
-        with open(file_path, 'r') as file:
-            return json.load(file)
-    except FileNotFoundError:
-        print(f"Error: Configuration file '{file_path}' not found.")
-        exit(1)
 
 # Function to fetch all CI/CD variables from a GitLab project
-def fetch_variables(config):
-    gitlab_url = config["gitlab_url"]
-    project_id = config["src_project_id"]
-    private_token = config["private_token"]
+def fetch_variables(project):
+    p_variables = list(project.variables.list(iterator=True))
+    variables = [var.asdict() for var in p_variables]
 
-    headers = {
-        "PRIVATE-TOKEN": private_token,
-        "Content-Type": "application/json"
-    }
+    return variables
 
-    url = f"{gitlab_url}/api/v4/projects/{project_id}/variables?per_page=100"
-    response = requests.get(url, headers=headers)
 
-    if response.status_code == 200:
-        variables = response.json()
-        return variables
-    else:
-        print(f"Failed to fetch variables. Status code: {response.status_code}")
-        return []
+def fetch_sched_variables(sched_pipeline):
+    variables = sched_pipeline.attributes["variables"]
+    return variables
 
-# Function to format the fetched variables for input into the config file
-def format_variables_for_input(variables):
-    ci_variables = {var['key']: var['value'] for var in variables}
-    return ci_variables
 
 # Main function to load the config and fetch variables
 def main():
     # Setup argument parser
-    parser = argparse.ArgumentParser(description="GitLab CI/CD Variables Reader")
+    parser = argparse.ArgumentParser(description="GitLab CI/CD Variable reader")
     parser.add_argument(
-        "--auth_config",
+        "--config_file",
         type=str,
-        default="auth-config.json",
-        help="Path to the configuration file (default: config.json)"
+        default="gitlab.ini",
+        required=True,
+        help="Path to the configuration file (default: gitlab.ini)",
+    )
+    parser.add_argument(
+        "--var_file",
+        type=str,
+        default="ci-variables.json",
+        help="Path to the CI vars file (default: ci-variables.json)",
+    )
+    parser.add_argument(
+        "--project_id",
+        type=int,
+        default="",
+        required=True,
+        help="Gitlab project ID to read variables from",
+    )
+    parser.add_argument(
+        "--sched_pipeline_id",
+        type=int,
+        help="Gitlab project scheduled pipeline ID",
     )
 
     # Parse the arguments
     args = parser.parse_args()
 
-    # Load the configuration file
-    config = load_config(args.auth_config)
+    gl = gitlab.Gitlab.from_config("uabrc", [args.config_file])
+    project = gl.projects.get(args.project_id)
 
-    # Fetch and print the variables
-    variables = fetch_variables(config)
-
-    if variables:
-        print("ci_variables:")
-        formatted_variables = format_variables_for_input(variables)
-        # Print the formatted variables dictionary in JSON format
-        print(json.dumps(formatted_variables, indent=4))
+    # Fetch project or sched pipeline variables
+    if not args.sched_pipeline_id:
+        variables = fetch_variables(project)
     else:
-        print("No variables found.")
+        sched_pipeline = project.pipelineschedules.get(args.sched_pipeline_id)
+        variables = fetch_sched_variables(sched_pipeline)
+
+    try:
+        with open(args.var_file, "w") as file:
+            json.dump(variables, file, indent=2)
+    except FileNotFoundError:
+        print(f"Error: Writing File to '{args.var_file}'")
+        exit(1)
+
 
 # Run the main function
 if __name__ == "__main__":
     main()
-

scripts/gitlab-ci-vars-updater.py

-import requests
-import json
 import argparse
+import json
+
+import gitlab
 
-# Load configuration from JSON file
-def load_config(file_path):
+
+def load_file(file_path):
     try:
-        with open(file_path, 'r') as file:
+        with open(file_path, "r") as file:
             return json.load(file)
     except FileNotFoundError:
         print(f"Error: Configuration file '{file_path}' not found.")
         exit(1)
 
+
 # Function to create or update a GitLab CI/CD variable
-def create_or_update_variable(config, var_name, var_value):
-    gitlab_url = config["gitlab_url"]
-    project_id = config["dest_project_id"]
-    private_token = config["private_token"]
-
-    headers = {
-        "PRIVATE-TOKEN": private_token,
-        "Content-Type": "application/json"
-    }
-
-    url = f"{gitlab_url}/api/v4/projects/{project_id}/variables/{var_name}"
-
-    # Check if the variable already exists
-    response = requests.get(url, headers=headers)
-
-    if response.status_code == 404:
-        print(f"Variable '{var_name}' does not exist. Creating new variable...")
-        create_url = f"{gitlab_url}/api/v4/projects/{project_id}/variables"
-        data = {
-            "key": var_name,
-            "value": var_value,
-            "variable_type": "env_var",  # Can be 'file' for file variables
-            "protected": False,  # Set to True if you want the variable protected
-            "masked": False  # Set to True if you want the variable masked
-        }
-        create_response = requests.post(create_url, headers=headers, json=data)
-
-        if create_response.status_code == 201:
-            print(f"Variable '{var_name}' created successfully.")
+def create_or_update_variable(project, var_dict):
+    # Check if the variable exists in the project
+    p_variable = project.variables.get(var_dict["key"]).asdict()
+    if p_variable:
+        if p_variable != var_dict:
+            # Check if the attributes are the same
+            for k, v in var_dict.items():
+                if p_variable[k] != v:
+                    # If not update the value in the project
+                    print(f"Updating key {k} value")
+                    project.variables.update(k, {"value": v})
         else:
-            print(f"Failed to create variable '{var_name}'. Status code: {create_response.status_code}")
-    elif response.status_code == 200:
-        print(f"Variable '{var_name}' already exists. Updating variable...")
-        data = {"value": var_value}
-        update_response = requests.put(url, headers=headers, json=data)
-
-        if update_response.status_code == 200:
-            print(f"Variable '{var_name}' updated successfully.")
+            print(f"variable {var_dict["key"]} already exists")
+    # Create variable if it doesn't exist in the project
+    else:
+        for k, v in var_dict.items():
+            print(f"Creating variable {var_dict["key"]}")
+            project.variables.create(var_dict)
+
+
+def get_pipeline_vars_by_key(sched_pipeline, key_name):
+    p_vars = sched_pipeline.attributes["variables"]
+    # p_vars.sort(key=lambda x: x["key"])
+    for p_variable in p_vars:
+        if p_variable.get("key") == key_name:
+            return p_variable
+
+
+# Function to create or update a schedule pipeline variable
+def create_or_update_sched_vars(sched_pipeline, var_dict):
+    # Check if the variable exists in the sched pipeline
+    p_variable = get_pipeline_vars_by_key(sched_pipeline, var_dict["key"])
+    if p_variable:
+        if p_variable != var_dict:
+            # Check if the attributes are the same
+            for k, v in var_dict.items():
+                if p_variable[k] != v:
+                    # If not update the value in the project
+                    print(f"Updating key {k} value")
+                    sched_pipeline.variables.delete(p_variable["key"])
+                    sched_pipeline.variables.create(var_dict)
         else:
-            print(f"Failed to update variable '{var_name}'. Status code: {update_response.status_code}")
+            print(f"variable {var_dict["key"]} already exists")
+    # Create variable if it doesn't exist in the project
     else:
-        print(f"Error checking variable '{var_name}'. Status code: {response.status_code}")
+        for k, v in var_dict.items():
+            print(f"Creating variable {var_dict["key"]}")
+            return sched_pipeline.variables.create(var_dict)
+
 
-# Main function to load the auth config and apply variables
 def main():
     # Setup argument parser
     parser = argparse.ArgumentParser(description="GitLab CI/CD Variables Updater")
     parser.add_argument(
-        "--auth_config",
+        "--config_file",
         type=str,
-        default="auth-config.json",
-        help="Path to the configuration file (default: config.json)"
+        default="gitlab.ini",
+        required=True,
+        help="Path to the configuration file (default: gitlab.ini)",
     )
     parser.add_argument(
-        "--vars",
+        "--var_file",
         type=str,
         default="ci-variables.json",
-        help="Path to the configuration file (default: config.json)"
+        help="Path to the CI vars file (default: ci-variables.json)",
+    )
+    parser.add_argument(
+        "--project_id",
+        type=int,
+        required=True,
+        help="Gitlab project ID for target variables",
+    )
+    parser.add_argument(
+        "--sched_pipeline_id",
+        type=int,
+        help="Gitlab project scheduled pipeline ID",
     )
 
     # Parse the arguments
     args = parser.parse_args()
 
-    # Load the configuration file
-    config = load_config(args.auth_config)
-    vars_dict = load_config(args.vars)
+    gl = gitlab.Gitlab.from_config("uabrc", [args.config_file])
+    project = gl.projects.get(args.project_id)
+
+    # Load the CI vars file
+    var_list = load_file(args.var_file)
+    # var_list.sort(key=lambda x: x["key"])
 
-    ci_variables = vars_dict["ci_variables"]
+    # Create or update all variables
+    for var_dict in var_list:
+        if not args.sched_pipeline_id:
+            create_or_update_variable(project, var_dict)
+        else:
+            sched_pipeline = project.pipelineschedules.get(args.sched_pipeline_id)
+            create_or_update_sched_vars(sched_pipeline, var_dict)
 
-    # Loop through and create/update all variables
-    for var_name, var_value in ci_variables.items():
-        create_or_update_variable(config, var_name, var_value)
 
-# Run the main function
 if __name__ == "__main__":
     main()
-