Skip to content

prefect_gitlab

GitLabCredentials

Bases: Block

Store a GitLab personal access token to interact with private GitLab repositories.

Attributes:

Name Type Description
token Optional[SecretStr]

The personal access token to authenticate with GitLab.
url Optional[str]

URL to self-hosted GitLab instances.

Examples:

Load stored GitLab credentials:

from prefect_gitlab import GitLabCredentials
gitlab_credentials_block = GitLabCredentials.load("BLOCK_NAME")
Source code in prefect_gitlab/credentials.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class GitLabCredentials(Block):
    """
    Store a GitLab personal access token to interact with private GitLab
    repositories.

    Attributes:
        token: The personal access token to authenticate with GitLab.
        url: URL to self-hosted GitLab instances.

    Examples:
        Load stored GitLab credentials:
        ```python
        from prefect_gitlab import GitLabCredentials
        gitlab_credentials_block = GitLabCredentials.load("BLOCK_NAME")
        ```
    """

    _block_type_name = "GitLab Credentials"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/55edIimT4g9gbjhkh5a3Sp/dfdb9391d8f45c2e93e72e3a4d350771/gitlab-logo-500.png?h=250"

    token: Optional[SecretStr] = Field(
        title="Personal Access Token",
        default=None,
        description="A GitLab Personal Access Token with read_repository scope.",
    )
    url: Optional[str] = Field(
        default=None, title="URL", description="URL to self-hosted GitLab instances."
    )

    def get_client(self) -> Gitlab:
        """
        Gets an authenticated GitLab client.

        Returns:
            An authenticated GitLab client.
        """
        # ref: https://python-gitlab.readthedocs.io/en/stable/
        gitlab = Gitlab(url=self.url, oauth_token=self.token.get_secret_value())
        gitlab.auth()
        return gitlab

get_client()

Gets an authenticated GitLab client.

Returns:

Type Description
Gitlab

An authenticated GitLab client.
Source code in prefect_gitlab/credentials.py 
40
41
42
43
44
45
46
47
48
49
50
def get_client(self) -> Gitlab:
    """
    Gets an authenticated GitLab client.

    Returns:
        An authenticated GitLab client.
    """
    # ref: https://python-gitlab.readthedocs.io/en/stable/
    gitlab = Gitlab(url=self.url, oauth_token=self.token.get_secret_value())
    gitlab.auth()
    return gitlab

GitLabRepository

Bases: ReadableDeploymentStorage

Interact with files stored in GitLab repositories.

An accessible installation of git is required for this block to function properly.

Source code in prefect_gitlab/repositories.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
class GitLabRepository(ReadableDeploymentStorage):
    """
    Interact with files stored in GitLab repositories.

    An accessible installation of git is required for this block to function
    properly.
    """

    _block_type_name = "GitLab Repository"
    _logo_url = "https://images.ctfassets.net/gm98wzqotmnx/55edIimT4g9gbjhkh5a3Sp/dfdb9391d8f45c2e93e72e3a4d350771/gitlab-logo-500.png?h=250"
    _description = "Interact with files stored in GitLab repositories."

    repository: str = Field(
        default=...,
        description=(
            "The URL of a GitLab repository to read from, in either HTTP/HTTPS or SSH format."  # noqa
        ),
    )
    reference: Optional[str] = Field(
        default=None,
        description="An optional reference to pin to; can be a branch name or tag.",
    )
    git_depth: Optional[int] = Field(
        default=1,
        gte=1,
        description="The number of commits that Git history is truncated to "
        "during cloning. Set to None to fetch the entire history.",
    )
    credentials: Optional[GitLabCredentials] = Field(
        default=None,
        description="An optional GitLab Credentials block for authenticating with "
        "private GitLab repos.",
    )

    def _create_repo_url(self) -> str:
        """Format the URL provided to the `git clone` command.
        For private repos: https://<oauth-key>@gitlab.com/<username>/<repo>.git
        All other repos should be the same as `self.repository`.
        """
        url_components = urllib.parse.urlparse(self.repository)
        if url_components.scheme in ["https", "http"] and self.credentials is not None:
            token = self.credentials.token.get_secret_value()
            updated_components = url_components._replace(
                netloc=f"oauth2:{token}@{url_components.netloc}"
            )
            full_url = urllib.parse.urlunparse(updated_components)
        else:
            full_url = self.repository

        return full_url

    @staticmethod
    def _get_paths(
        dst_dir: Union[str, None], src_dir: str, sub_directory: Optional[str]
    ) -> Tuple[str, str]:
        """Returns the fully formed paths for GitLabRepository contents in the form
        (content_source, content_destination).
        """
        if dst_dir is None:
            content_destination = Path(".").absolute()
        else:
            content_destination = Path(dst_dir)

        content_source = Path(src_dir)

        if sub_directory:
            content_destination = content_destination.joinpath(sub_directory)
            content_source = content_source.joinpath(sub_directory)

        return str(content_source), str(content_destination)

    @sync_compatible
    @retry(
        stop=stop_after_attempt(MAX_CLONE_ATTEMPTS),
        wait=wait_fixed(CLONE_RETRY_MIN_DELAY_SECONDS)
        + wait_random(
            CLONE_RETRY_MIN_DELAY_JITTER_SECONDS,
            CLONE_RETRY_MAX_DELAY_JITTER_SECONDS,
        ),
        reraise=True,
    )
    async def get_directory(
        self, from_path: Optional[str] = None, local_path: Optional[str] = None
    ) -> None:
        """
        Clones a GitLab project specified in `from_path` to the provided `local_path`;
        defaults to cloning the repository reference configured on the Block to the
        present working directory.
        Args:
            from_path: If provided, interpreted as a subdirectory of the underlying
                repository that will be copied to the provided local path.
            local_path: A local path to clone to; defaults to present working directory.
        """
        # CONSTRUCT COMMAND
        cmd = ["git", "clone", self._create_repo_url()]
        if self.reference:
            cmd += ["-b", self.reference]

        # Limit git history
        if self.git_depth is not None:
            cmd += ["--depth", f"{self.git_depth}"]

        # Clone to a temporary directory and move the subdirectory over
        with TemporaryDirectory(suffix="prefect") as tmp_dir:
            cmd.append(tmp_dir)

            err_stream = io.StringIO()
            out_stream = io.StringIO()
            process = await run_process(cmd, stream_output=(out_stream, err_stream))
            if process.returncode != 0:
                err_stream.seek(0)
                raise OSError(f"Failed to pull from remote:\n {err_stream.read()}")

            content_source, content_destination = self._get_paths(
                dst_dir=local_path, src_dir=tmp_dir, sub_directory=from_path
            )

            shutil.copytree(
                src=content_source, dst=content_destination, dirs_exist_ok=True
            )

get_directory(from_path=None, local_path=None) async

Clones a GitLab project specified in from_path to the provided local_path; defaults to cloning the repository reference configured on the Block to the present working directory. Args: from_path: If provided, interpreted as a subdirectory of the underlying repository that will be copied to the provided local path. local_path: A local path to clone to; defaults to present working directory.

Source code in prefect_gitlab/repositories.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
@sync_compatible
@retry(
    stop=stop_after_attempt(MAX_CLONE_ATTEMPTS),
    wait=wait_fixed(CLONE_RETRY_MIN_DELAY_SECONDS)
    + wait_random(
        CLONE_RETRY_MIN_DELAY_JITTER_SECONDS,
        CLONE_RETRY_MAX_DELAY_JITTER_SECONDS,
    ),
    reraise=True,
)
async def get_directory(
    self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
    """
    Clones a GitLab project specified in `from_path` to the provided `local_path`;
    defaults to cloning the repository reference configured on the Block to the
    present working directory.
    Args:
        from_path: If provided, interpreted as a subdirectory of the underlying
            repository that will be copied to the provided local path.
        local_path: A local path to clone to; defaults to present working directory.
    """
    # CONSTRUCT COMMAND
    cmd = ["git", "clone", self._create_repo_url()]
    if self.reference:
        cmd += ["-b", self.reference]

    # Limit git history
    if self.git_depth is not None:
        cmd += ["--depth", f"{self.git_depth}"]

    # Clone to a temporary directory and move the subdirectory over
    with TemporaryDirectory(suffix="prefect") as tmp_dir:
        cmd.append(tmp_dir)

        err_stream = io.StringIO()
        out_stream = io.StringIO()
        process = await run_process(cmd, stream_output=(out_stream, err_stream))
        if process.returncode != 0:
            err_stream.seek(0)
            raise OSError(f"Failed to pull from remote:\n {err_stream.read()}")

        content_source, content_destination = self._get_paths(
            dst_dir=local_path, src_dir=tmp_dir, sub_directory=from_path
        )

        shutil.copytree(
            src=content_source, dst=content_destination, dirs_exist_ok=True
        )