Gitlab

zenml.integrations.gitlab special

Initialization of the GitLab ZenML integration.

GitLabIntegration (Integration)

Definition of GitLab integration for ZenML.

Source code in zenml/integrations/gitlab/__init__.py

class GitLabIntegration(Integration):
    """Definition of GitLab integration for ZenML."""

    NAME = GITLAB
    REQUIREMENTS: List[str] = ["python-gitlab"]

code_repositories special

Initialization of the ZenML GitLab code repository.

gitlab_code_repository

GitLab code repository.

GitLabCodeRepository (BaseCodeRepository)

GitLab code repository.

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

class GitLabCodeRepository(BaseCodeRepository):
    """GitLab code repository."""

    @property
    def config(self) -> GitLabCodeRepositoryConfig:
        """Returns the `GitLabCodeRepositoryConfig` config.

        Returns:
            The configuration.
        """
        return GitLabCodeRepositoryConfig(**self._config)

    @property
    def gitlab_project(self) -> Project:
        """The GitLab project object from the GitLab API.

        Returns:
            The GitLab project object.
        """
        return self._gitlab_session.projects.get(
            f"{self.config.group}/{self.config.project}"
        )

    def login(self) -> None:
        """Logs in to GitLab.

        Raises:
            RuntimeError: If the login fails.
        """
        try:
            self._gitlab_session = Gitlab(
                self.config.url, private_token=self.config.token
            )
            self._gitlab_session.auth()
            user = self._gitlab_session.user or None
            if user:
                logger.debug(f"Logged in as {user.username}")
        except Exception as e:
            raise RuntimeError(f"An error occurred while logging in: {str(e)}")

    def download_files(
        self, commit: str, directory: str, repo_sub_directory: Optional[str]
    ) -> None:
        """Downloads files from a commit to a local directory.

        Args:
            commit: The commit to download.
            directory: The directory to download to.
            repo_sub_directory: The sub directory to download from.
        """
        contents = self.gitlab_project.repository_tree(
            ref=commit,
            path=repo_sub_directory or "",
        )
        for content in contents:
            logger.debug(f"Processing {content['path']}")
            if content["type"] == "tree":
                path = os.path.join(directory, content["name"])
                os.makedirs(path, exist_ok=True)
                self.download_files(
                    commit=commit,
                    directory=path,
                    repo_sub_directory=content["path"],
                )
            else:
                try:
                    path = content["path"]
                    file_content = self.gitlab_project.files.get(
                        file_path=path, ref=commit
                    ).decode()
                    path = os.path.join(directory, content["name"])
                    with open(path, "wb") as file:
                        file.write(file_content)
                except Exception as e:
                    logger.error("Error processing %s: %s", content["path"], e)

    def get_local_context(self, path: str) -> Optional[LocalRepositoryContext]:
        """Gets the local repository context.

        Args:
            path: The path to the local repository.

        Returns:
            The local repository context.
        """
        return LocalGitRepositoryContext.at(
            path=path,
            code_repository_id=self.id,
            remote_url_validation_callback=self.check_remote_url,
        )

    def check_remote_url(self, url: str) -> bool:
        """Checks whether the remote url matches the code repository.

        Args:
            url: The remote url.

        Returns:
            Whether the remote url is correct.
        """
        https_url = f"https://{self.config.host}/{self.config.group}/{self.config.project}.git"
        if url == https_url:
            return True

        ssh_regex = re.compile(
            f".*{self.config.host}:{self.config.group}/{self.config.project}.git"
        )
        if ssh_regex.fullmatch(url):
            return True

        return False

config: GitLabCodeRepositoryConfig property readonly

Returns the GitLabCodeRepositoryConfig config.

Returns:

Type	Description
GitLabCodeRepositoryConfig	The configuration.

gitlab_project: gitlab.v4.objects.Project property readonly

The GitLab project object from the GitLab API.

Returns:

Type	Description
gitlab.v4.objects.Project	The GitLab project object.

check_remote_url(self, url)

Checks whether the remote url matches the code repository.

Parameters:

Name	Type	Description	Default
url	str	The remote url.	required

Returns:

Type	Description
bool	Whether the remote url is correct.

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

def check_remote_url(self, url: str) -> bool:
    """Checks whether the remote url matches the code repository.

    Args:
        url: The remote url.

    Returns:
        Whether the remote url is correct.
    """
    https_url = f"https://{self.config.host}/{self.config.group}/{self.config.project}.git"
    if url == https_url:
        return True

    ssh_regex = re.compile(
        f".*{self.config.host}:{self.config.group}/{self.config.project}.git"
    )
    if ssh_regex.fullmatch(url):
        return True

    return False

download_files(self, commit, directory, repo_sub_directory)

Downloads files from a commit to a local directory.

Parameters:

Name	Type	Description	Default
commit	str	The commit to download.	required
directory	str	The directory to download to.	required
repo_sub_directory	Optional[str]	The sub directory to download from.	required

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

def download_files(
    self, commit: str, directory: str, repo_sub_directory: Optional[str]
) -> None:
    """Downloads files from a commit to a local directory.

    Args:
        commit: The commit to download.
        directory: The directory to download to.
        repo_sub_directory: The sub directory to download from.
    """
    contents = self.gitlab_project.repository_tree(
        ref=commit,
        path=repo_sub_directory or "",
    )
    for content in contents:
        logger.debug(f"Processing {content['path']}")
        if content["type"] == "tree":
            path = os.path.join(directory, content["name"])
            os.makedirs(path, exist_ok=True)
            self.download_files(
                commit=commit,
                directory=path,
                repo_sub_directory=content["path"],
            )
        else:
            try:
                path = content["path"]
                file_content = self.gitlab_project.files.get(
                    file_path=path, ref=commit
                ).decode()
                path = os.path.join(directory, content["name"])
                with open(path, "wb") as file:
                    file.write(file_content)
            except Exception as e:
                logger.error("Error processing %s: %s", content["path"], e)

get_local_context(self, path)

Gets the local repository context.

Parameters:

Name	Type	Description	Default
path	str	The path to the local repository.	required

Returns:

Type	Description
Optional[zenml.code_repositories.local_repository_context.LocalRepositoryContext]	The local repository context.

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

def get_local_context(self, path: str) -> Optional[LocalRepositoryContext]:
    """Gets the local repository context.

    Args:
        path: The path to the local repository.

    Returns:
        The local repository context.
    """
    return LocalGitRepositoryContext.at(
        path=path,
        code_repository_id=self.id,
        remote_url_validation_callback=self.check_remote_url,
    )

login(self)

Logs in to GitLab.

Exceptions:

Type	Description
RuntimeError	If the login fails.

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

def login(self) -> None:
    """Logs in to GitLab.

    Raises:
        RuntimeError: If the login fails.
    """
    try:
        self._gitlab_session = Gitlab(
            self.config.url, private_token=self.config.token
        )
        self._gitlab_session.auth()
        user = self._gitlab_session.user or None
        if user:
            logger.debug(f"Logged in as {user.username}")
    except Exception as e:
        raise RuntimeError(f"An error occurred while logging in: {str(e)}")

GitLabCodeRepositoryConfig (BaseCodeRepositoryConfig)

Config for GitLab code repositories.

Parameters:

Name	Type	Description	Default
url		The full URL of the GitLab project.	required
group		The group of the project.	required
project		The name of the GitLab project.	required
host		The host of GitLab in case it is self-hosted instance.	required
token		The token to access the repository.	required

Source code in zenml/integrations/gitlab/code_repositories/gitlab_code_repository.py

class GitLabCodeRepositoryConfig(BaseCodeRepositoryConfig):
    """Config for GitLab code repositories.

    Args:
        url: The full URL of the GitLab project.
        group: The group of the project.
        project: The name of the GitLab project.
        host: The host of GitLab in case it is self-hosted instance.
        token: The token to access the repository.
    """

    url: Optional[str]
    group: str
    project: str
    host: Optional[str] = "gitlab.com"
    token: str = SecretField()