Skip to content

alternate_paths#

Textual's Implementation of Pathlib, Powered by fsspec

GitHubTextualPath #

Bases: GitHubPath

GitHubPath

UPath implementation for GitHub to be compatible with the Directory Tree

Source code in textual_universal_directorytree/alternate_paths.py 
 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
 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
class GitHubTextualPath(GitHubPath):
    """
    GitHubPath

    UPath implementation for GitHub to be compatible with the Directory Tree
    """

    @classmethod
    def _transform_init_args(
        cls,
        args: tuple[str | os.PathLike[Any], ...],
        protocol: str,
        storage_options: dict[str, Any],
    ) -> tuple[tuple[str | os.PathLike[Any], ...], str, dict[str, Any]]:
        """
        Initialize the GitHubPath with GitHub Token Authentication
        """
        if "token" not in storage_options:
            token = os.getenv("GITHUB_TOKEN")
            if token is not None:
                storage_options.update({"username": "Bearer", "token": token})
        handled_args = args
        if "sha" not in storage_options:
            handled_url = cls.handle_github_url(args[0])
            handled_args = (handled_url, *args[1:])
        return handled_args, protocol, storage_options

    def __str__(self) -> str:
        """
        String representation of the GitHubPath
        """
        return (
            f"{self.protocol}://{self.storage_options['org']}:"
            f"{self.storage_options['repo']}@{self.storage_options['sha']}/"
            f"{self.path}"
        )

    @classmethod
    def handle_github_url(cls, url: str | GitHubPath) -> str:
        """
        Handle GitHub URLs

        GitHub URLs are handled by converting them to the raw URL.
        """
        try:
            import requests
        except ImportError as e:
            raise ImportError(
                "The requests library is required to browse GitHub files. "
                "Install `textual-universal-directorytree` with the `remote` "
                "extra to install requests."
            ) from e
        url = str(url)
        gitub_prefix = "github://"
        if gitub_prefix in url and "@" not in url:
            _, user_password = url.split("github://")
            org, repo_str = user_password.split(":")
            repo, *args = repo_str.split("/")
        elif gitub_prefix in url and "@" in url:
            return url
        else:
            msg = f"Invalid GitHub URL: {url}"
            raise ValueError(msg)
        token = os.getenv("GITHUB_TOKEN")
        auth = {"auth": ("Bearer", token)} if token is not None else {}
        resp = requests.get(
            f"https://api.github.com/repos/{org}/{repo}",
            headers={"Accept": "application/vnd.github.v3+json"},
            timeout=30,
            **auth,  # type: ignore[arg-type]
        )
        resp.raise_for_status()
        default_branch = resp.json()["default_branch"]
        arg_str = "/".join(args)
        github_uri = f"{gitub_prefix}{org}:{repo}@{default_branch}/{arg_str}"
        return github_uri

    @property
    def name(self) -> str:
        """
        Override the name for top level repo
        """
        original_name = super().name
        if original_name == "":
            return self.storage_options["repo"]
        else:
            return original_name

name: str property #

Override the name for top level repo

__str__() #

String representation of the GitHubPath

Source code in textual_universal_directorytree/alternate_paths.py 
43
44
45
46
47
48
49
50
51
def __str__(self) -> str:
    """
    String representation of the GitHubPath
    """
    return (
        f"{self.protocol}://{self.storage_options['org']}:"
        f"{self.storage_options['repo']}@{self.storage_options['sha']}/"
        f"{self.path}"
    )

handle_github_url(url) classmethod #

Handle GitHub URLs

GitHub URLs are handled by converting them to the raw URL.

Source code in textual_universal_directorytree/alternate_paths.py 
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
@classmethod
def handle_github_url(cls, url: str | GitHubPath) -> str:
    """
    Handle GitHub URLs

    GitHub URLs are handled by converting them to the raw URL.
    """
    try:
        import requests
    except ImportError as e:
        raise ImportError(
            "The requests library is required to browse GitHub files. "
            "Install `textual-universal-directorytree` with the `remote` "
            "extra to install requests."
        ) from e
    url = str(url)
    gitub_prefix = "github://"
    if gitub_prefix in url and "@" not in url:
        _, user_password = url.split("github://")
        org, repo_str = user_password.split(":")
        repo, *args = repo_str.split("/")
    elif gitub_prefix in url and "@" in url:
        return url
    else:
        msg = f"Invalid GitHub URL: {url}"
        raise ValueError(msg)
    token = os.getenv("GITHUB_TOKEN")
    auth = {"auth": ("Bearer", token)} if token is not None else {}
    resp = requests.get(
        f"https://api.github.com/repos/{org}/{repo}",
        headers={"Accept": "application/vnd.github.v3+json"},
        timeout=30,
        **auth,  # type: ignore[arg-type]
    )
    resp.raise_for_status()
    default_branch = resp.json()["default_branch"]
    arg_str = "/".join(args)
    github_uri = f"{gitub_prefix}{org}:{repo}@{default_branch}/{arg_str}"
    return github_uri

S3TextualPath #

Bases: S3Path

S3TextualPath

Source code in textual_universal_directorytree/alternate_paths.py 
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
class S3TextualPath(S3Path):
    """
    S3TextualPath
    """

    def _is_top_level_bucket(self) -> bool:
        """
        Check if the path is a top level bucket
        """
        return len(self.parts) == 1 and self.parts[0] == "/"

    def __str__(self) -> str:
        """
        String representation of the S3Path
        """
        return re.sub(r"s3:/*", "s3://", super().__str__())

    @property
    def name(self) -> str:
        """
        Override the name for top level repo
        """
        original_name = super().name
        if self._is_top_level_bucket():
            return f"{self._url.scheme}://{self._url.netloc}"
        elif original_name == "":
            return self.parts[-1].rstrip("/")
        else:
            return original_name

name: str property #

Override the name for top level repo

__str__() #

String representation of the S3Path

Source code in textual_universal_directorytree/alternate_paths.py 
116
117
118
119
120
def __str__(self) -> str:
    """
    String representation of the S3Path
    """
    return re.sub(r"s3:/*", "s3://", super().__str__())

SFTPTextualPath #

Bases: UPath

SFTPTextualPath

Source code in textual_universal_directorytree/alternate_paths.py 
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
class SFTPTextualPath(UPath):
    """
    SFTPTextualPath
    """

    @property
    def path(self) -> str:
        """
        Always return the path relative to the root
        """
        pth = super().path
        if pth.startswith("."):
            return f"/{pth[1:]}"
        elif pth.startswith("/"):
            return pth
        else:
            return "/" + pth

    def __str__(self) -> str:
        """
        Add the protocol prefix + extras to the string representation
        """
        string_representation = f"{self.protocol}://"
        if "username" in self.storage_options:
            string_representation += f"{self.storage_options['username']}@"
        string_representation += f"{self.storage_options['host']}"
        if "port" in self.storage_options:
            string_representation += f":{self.storage_options['port']}"
        string_representation += self.path
        return string_representation

    @property
    def name(self) -> str:
        """
        Override the name for top level repo
        """
        original_name = super().name
        if original_name == "":
            return "/"
        else:
            return original_name

name: str property #

Override the name for top level repo

path: str property #

Always return the path relative to the root

__str__() #

Add the protocol prefix + extras to the string representation

Source code in textual_universal_directorytree/alternate_paths.py 
154
155
156
157
158
159
160
161
162
163
164
165
def __str__(self) -> str:
    """
    Add the protocol prefix + extras to the string representation
    """
    string_representation = f"{self.protocol}://"
    if "username" in self.storage_options:
        string_representation += f"{self.storage_options['username']}@"
    string_representation += f"{self.storage_options['host']}"
    if "port" in self.storage_options:
        string_representation += f":{self.storage_options['port']}"
    string_representation += self.path
    return string_representation