Skip to content

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) pydantic-model

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()