Skip to content

prefect_github

GitHubCredentials

Bases: CredentialsBlock

Block used to manage GitHub authentication.

Attributes:

Name Type Description
token Optional[SecretStr]

the token to authenticate into GitHub.

Examples:

Load stored GitHub credentials:

from prefect_github import GitHubCredentials
github_credentials_block = GitHubCredentials.load("BLOCK_NAME")
Source code in prefect_github/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
51
52
53
54
55
56
57
58
59
class GitHubCredentials(CredentialsBlock):
    """
    Block used to manage GitHub authentication.

    Attributes:
        token: the token to authenticate into GitHub.

    Examples:
        Load stored GitHub credentials:
        ```python
        from prefect_github import GitHubCredentials
        github_credentials_block = GitHubCredentials.load("BLOCK_NAME")
        ```
    """

    _block_type_name = "GitHub Credentials"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/41971cfecfea5f79ff334164f06ecb34d1038dd4-250x250.png"  # noqa
    _documentation_url = "https://prefecthq.github.io/prefect-github/credentials/#prefect_github.credentials.GitHubCredentials"  # noqa

    token: Optional[SecretStr] = Field(
        default=None, description="A GitHub personal access token (PAT)."
    )

    def get_client(self) -> HTTPEndpoint:
        """
        Gets an authenticated GitHub GraphQL HTTPEndpoint client.

        Returns:
            An authenticated GitHub GraphQL HTTPEndpoint client.

        Example:
            Gets an authenticated GitHub GraphQL HTTPEndpoint client.
            ```python
            from prefect_github import GitHubCredentials

            github_credentials = GitHubCredentials(token=token)
            client = github_credentials.get_client()
            ```
        """

        if self.token is not None:
            base_headers = {"Authorization": f"Bearer {self.token.get_secret_value()}"}
        else:
            base_headers = None

        endpoint = HTTPEndpoint(
            "https://api.github.com/graphql", base_headers=base_headers
        )
        return endpoint

get_client()

Gets an authenticated GitHub GraphQL HTTPEndpoint client.

Returns:

Type Description
HTTPEndpoint

An authenticated GitHub GraphQL HTTPEndpoint client.
Example

Gets an authenticated GitHub GraphQL HTTPEndpoint client.

from prefect_github import GitHubCredentials

github_credentials = GitHubCredentials(token=token)
client = github_credentials.get_client()
Source code in prefect_github/credentials.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def get_client(self) -> HTTPEndpoint:
    """
    Gets an authenticated GitHub GraphQL HTTPEndpoint client.

    Returns:
        An authenticated GitHub GraphQL HTTPEndpoint client.

    Example:
        Gets an authenticated GitHub GraphQL HTTPEndpoint client.
        ```python
        from prefect_github import GitHubCredentials

        github_credentials = GitHubCredentials(token=token)
        client = github_credentials.get_client()
        ```
    """

    if self.token is not None:
        base_headers = {"Authorization": f"Bearer {self.token.get_secret_value()}"}
    else:
        base_headers = None

    endpoint = HTTPEndpoint(
        "https://api.github.com/graphql", base_headers=base_headers
    )
    return endpoint

GitHubRepository

Bases: ReadableDeploymentStorage

Interact with files stored on GitHub repositories.

Source code in prefect_github/repository.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
class GitHubRepository(ReadableDeploymentStorage):
    """
    Interact with files stored on GitHub repositories.
    """

    _block_type_name = "GitHub Repository"
    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/41971cfecfea5f79ff334164f06ecb34d1038dd4-250x250.png"  # noqa: E501
    _documentation_url = "https://prefecthq.github.io/prefect-github/repository/#prefect_github.repository.GitHubRepository"  # noqa

    repository_url: str = Field(
        default=...,
        title="Repository URL",
        description=(
            "The URL of a GitHub repository to read from, in either HTTPS or SSH "
            "format. If you are using a private repo, it must be in the HTTPS format."
        ),
    )
    reference: Optional[str] = Field(
        default=None,
        description="An optional reference to pin to; can be a branch name or tag.",
    )
    credentials: Optional[GitHubCredentials] = Field(
        default=None,
        description="An optional GitHubCredentials block for using private GitHub repos.",  # noqa: E501
    )

    def _create_repo_url(self) -> str:
        """Format the URL provided to the `git clone` command.

        For private repos: https://<oauth-key>@github.com/<username>/<repo>.git
        All other repos should be the same as `self.repository`.
        """
        url_components = urlparse(self.repository_url)
        if url_components.scheme == "https" and self.credentials is not None:
            token_value = self.credentials.token.get_secret_value()
            updated_components = url_components._replace(
                netloc=f"{token_value}@{url_components.netloc}"
            )
            full_url = urlunparse(updated_components)
        else:
            full_url = self.repository_url

        return full_url

    @staticmethod
    def _get_paths(
        dst_dir: Union[str, None], src_dir: str, sub_directory: str
    ) -> Tuple[str, str]:
        """Returns the fully formed paths for GitHubRepository 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
    async def get_directory(
        self, from_path: Optional[str] = None, local_path: Optional[str] = None
    ) -> None:
        """
        Clones a GitHub 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 = f"git clone {self._create_repo_url()}"
        if self.reference:
            cmd += f" -b {self.reference}"

        # Limit git history
        cmd += " --depth 1"

        # Clone to a temporary directory and move the subdirectory over
        with TemporaryDirectory(suffix="prefect") as tmp_dir:
            tmp_path_str = tmp_dir
            cmd += f" {tmp_path_str}"
            cmd = shlex.split(cmd)

            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 RuntimeError(f"Failed to pull from remote:\n {err_stream.read()}")

            content_source, content_destination = self._get_paths(
                dst_dir=local_path, src_dir=tmp_path_str, 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 GitHub 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.

Parameters:

Name Type Description Default
from_path Optional[str]

If provided, interpreted as a subdirectory of the underlying repository that will be copied to the provided local path.

 None
local_path Optional[str]

A local path to clone to; defaults to present working directory.

 None
Source code in prefect_github/repository.py 
 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
@sync_compatible
async def get_directory(
    self, from_path: Optional[str] = None, local_path: Optional[str] = None
) -> None:
    """
    Clones a GitHub 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 = f"git clone {self._create_repo_url()}"
    if self.reference:
        cmd += f" -b {self.reference}"

    # Limit git history
    cmd += " --depth 1"

    # Clone to a temporary directory and move the subdirectory over
    with TemporaryDirectory(suffix="prefect") as tmp_dir:
        tmp_path_str = tmp_dir
        cmd += f" {tmp_path_str}"
        cmd = shlex.split(cmd)

        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 RuntimeError(f"Failed to pull from remote:\n {err_stream.read()}")

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

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