Nexus utilitary for Maven projects (requires requests and lxml).

maven_nexus.py

import os, re, logging
import lxml.etree as ET

import requests
from requests.auth import HTTPBasicAuth


class Nexus:

    def __init__(self):
        maven_settings_path = os.getenv('MAVEN_SETTINGS_PATH', f"{os.getenv('HOME')}/.m2/settings.xml")
        if os.path.exists(maven_settings_path):
            tree = ET.parse(maven_settings_path)
            root = tree.getroot()
            # Find the mirror URL and credentials in the settings.xml file that has namespace on top
            self.url = root.find(".//{*}mirror/{*}url").text
            username = root.find(".//{*}server/{*}username").text
            password = root.find(".//{*}server/{*}password").text
            pattern = re.compile("^\\$\\{env\\..*\\}$")
            if pattern.match(username):
                username = os.getenv(username[6:-1])
            if pattern.match(password):
                password = os.getenv(password[6:-1])
        else:
            self.url = os.getenv('NEXUS_URL')
            username = os.getenv('NEXUS_USERNAME')
            password = os.getenv('NEXUS_PASSWORD')
        if not self.url or not username or not password:
            logging.error("Nexus URL, username, and password must be provided.")
            raise ValueError("Nexus URL, username, and password must be provided.")
        self.auth = HTTPBasicAuth(username, password)

    def fetch_maven_metadata(self, group_id: str, artifact_id: str) -> ET.Element:
        """
        Fetch the Maven metadata for a given artifact from Nexus repository.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :return: An ElementTree Element representing the Maven metadata, or None if not found.
        """
        if not group_id or not artifact_id:
            logging.error("Group ID and Artifact ID must be provided.")
            return None
        url = f"{self.url}/{group_id.replace('.', '/')}/{artifact_id}/maven-metadata.xml"
        response = requests.get(url, auth=self.auth)
        if response.status_code == 200:
            maven_metadata_root = ET.fromstring(response.content)
            return maven_metadata_root
        else:
            logging.warning(f"Error fetching maven-metadata.xml: {url} ({response.status_code})")
            return None

    def get_latest_release_version(self, group_id: str, artifact_id: str) -> str:
        """
        Fetch the latest release version of an artifact from Nexus repository.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :return: The latest release version as a string, or None if not found.
        """
        maven_metadata_root = self.fetch_maven_metadata(group_id, artifact_id)
        if maven_metadata_root is None:
            return None

        versioning = maven_metadata_root.find("{*}versioning")
        if versioning is not None:
            latest_release_version = versioning.find("{*}release")
            if latest_release_version is not None:
                return latest_release_version.text
            else:
                logging.warning("No latest release version found in maven-metadata.xml.")
                return None
        else:
            logging.warning("No versioning information found in maven-metadata.xml.")
            return None

    def get_versions(self, group_id: str, artifact_id: str) -> list:
        """
        Fetch all versions of an artifact from Nexus repository.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :return: A list of version strings, or an empty list if not found.
        """
        maven_metadata_root = self.fetch_maven_metadata(group_id, artifact_id)
        if maven_metadata_root is None:
            return []

        versions = []
        versioning = maven_metadata_root.find("{*}versioning")
        if versioning is not None:
            for version in versioning.findall("{*}versions/{*}version"):
                versions.append(version.text)
        else:
            logging.warning("No versioning information found in maven-metadata.xml.")
        return versions

    def fetch_pom(self, group_id: str, artifact_id: str, version: str) -> ET.Element:
        """
        Fetch the POM file for a given artifact from Nexus repository.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :param version: The version of the artifact.
        :return: An ElementTree Element representing the POM file, or None if not found.
        """
        if not group_id or not artifact_id or not version:
            logging.error("Group ID, Artifact ID, and Version must be provided.")
            return None
        url = f"{self.url}/{group_id.replace('.', '/')}/{artifact_id}/{version}/{artifact_id}-{version}.pom"
        response = requests.get(url, auth=self.auth)
        if response.status_code == 200:
            pom_root = ET.fromstring(response.content)
            return pom_root
        else:
            logging.warning(f"Error fetching POM file: {url} ({response.status_code})")
            return None

    def get_dependencies(self, group_id: str, artifact_id: str, version: str) -> list:
        """
        Get the dependencies of an artifact from its POM file.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :param version: The version of the artifact.
        :return: A list of dependencies, each represented as a dictionary with 'groupId', 'artifactId', and 'version'.
        """
        pom_root = self.fetch_pom(group_id, artifact_id, version)
        if pom_root is None:
            return []

        dependencies = []
        for dependency in pom_root.findall("{*}dependencies/{*}dependency"):
            dep_group_id = dependency.find("{*}groupId")
            dep_artifact_id = dependency.find("{*}artifactId")
            dep_version = dependency.find("{*}version")
            if dep_group_id is not None and dep_artifact_id is not None and dep_version is not None:
                dependencies.append({
                    "groupId": dep_group_id.text,
                    "artifactId": dep_artifact_id.text,
                    "version": dep_version.text
                })
        return dependencies

    def get_dependencies_same_groupId(self, group_id: str, artifact_id: str, version: str) -> list:
        """
        Get the dependencies of an artifact that have the same group ID.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :param version: The version of the artifact.
        :return: A list of dependencies with the same group ID.
        """
        dependencies = self.get_dependencies(group_id, artifact_id, version)
        return [dep for dep in dependencies if dep['groupId'] == group_id]

    def get_modules(self, group_id: str, artifact_id: str, version: str) -> list:
        """
        Get the modules of a multi-module project from its POM file.
        :param group_id: The group ID of the artifact.
        :param artifact_id: The artifact ID of the artifact.
        :param version: The version of the artifact.
        :return: A list of module names.
        """
        pom_root = self.fetch_pom(group_id, artifact_id, version)
        if pom_root is None:
            return []

        modules = []
        for module in pom_root.findall("{*}modules/{*}module"):
            modules.append(module.text)
        return modules