Skip to content

Class swarmauri_gitfilter_gh_release.gh_release_filter.GithubReleaseFilter

swarmauri_gitfilter_gh_release.gh_release_filter.GithubReleaseFilter 

GithubReleaseFilter(
    token,
    org,
    repo,
    tag,
    *,
    release_name=None,
    message="",
    draft=False,
    prerelease=False,
    prefix="",
    **kwargs,
)

Bases: StorageAdapterBase, GitFilterBase

Git filter that uses GitHub Releases to store and retrieve assets.

Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
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
def __init__(
    self,
    token: SecretStr | str,
    org: str,
    repo: str,
    tag: str,
    *,
    release_name: Optional[str] = None,
    message: str = "",
    draft: bool = False,
    prerelease: bool = False,
    prefix: str = "",
    **kwargs,
) -> None:
    super().__init__(**kwargs)
    token_val = token.get_secret_value() if isinstance(token, SecretStr) else token
    self._client = Github(token_val)
    self._repo = self._client.get_organization(org).get_repo(repo)
    self._tag = tag
    self._prefix = prefix.lstrip("/")
    self._release = self._get_or_create_release(
        tag=tag,
        name=release_name or tag,
        message=message,
        draft=draft,
        prerelease=prerelease,
    )

root_uri property 

root_uri

upload 

upload(key, data)
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def upload(self, key: str, data: BinaryIO) -> str:
    key = self._full_key(key)
    data.seek(0)
    content = data.read()

    for asset in self._release.get_assets():
        if asset.name == key:
            asset.delete_asset()
            break

    with tempfile.NamedTemporaryFile() as tmp:
        tmp.write(content)
        tmp.flush()
        self._release.upload_asset(path=tmp.name, name=key, label=key)

    return f"{self.root_uri}{key.lstrip('/')}"

download 

download(key)
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
def download(self, key: str) -> BinaryIO:
    key = self._full_key(key)
    for asset in self._release.get_assets():
        if asset.name == key:
            _, raw = self._client._Github__requester.requestBytes(
                "GET",
                asset.url,
                headers={"Accept": "application/octet-stream"},
            )
            buf = io.BytesIO(raw)
            buf.seek(0)
            return buf
    raise FileNotFoundError(f"Asset '{key}' not found in release '{self._tag}'")

upload_dir 

upload_dir(src, *, prefix='')
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
113
114
115
116
117
118
119
120
def upload_dir(self, src: str | os.PathLike, *, prefix: str = "") -> None:
    base = Path(src)
    for path in base.rglob("*"):
        if path.is_file():
            rel = path.relative_to(base).as_posix()
            key = f"{prefix.rstrip('/')}/{rel}" if prefix else rel
            with path.open("rb") as fh:
                self.upload(key, fh)

iter_prefix 

iter_prefix(prefix)
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
122
123
124
125
126
127
128
129
130
def iter_prefix(self, prefix: str):
    full = self._full_key(prefix)
    for asset in self._release.get_assets():
        name = asset.name
        if name.startswith(full.rstrip("/")):
            key = name
            if self._prefix and key.startswith(self._prefix.rstrip("/") + "/"):
                key = key[len(self._prefix.rstrip("/")) + 1 :]
            yield key

download_prefix 

download_prefix(prefix, dest_dir)
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
132
133
134
135
136
137
138
139
def download_prefix(self, prefix: str, dest_dir: str | os.PathLike) -> None:
    dest = Path(dest_dir)
    for rel_key in self.iter_prefix(prefix):
        target = dest / rel_key
        target.parent.mkdir(parents=True, exist_ok=True)
        data = self.download(rel_key)
        with target.open("wb") as fh:
            shutil.copyfileobj(data, fh)

from_uri classmethod 

from_uri(uri)
Source code in swarmauri_gitfilter_gh_release/gh_release_filter.py 
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
@classmethod
def from_uri(cls, uri: str) -> "GithubReleaseFilter":
    from urllib.parse import urlparse

    p = urlparse(uri)
    org = p.netloc
    repo, tag, *rest = p.path.lstrip("/").split("/", 2)
    prefix = rest[0] if rest else ""

    cfg = load_peagen_toml()
    gh_cfg = cfg.get("storage", {}).get("filters", {}).get("gh_release", {})

    token = gh_cfg.get("token") or os.getenv("GITHUB_TOKEN", "")

    return cls(
        token=token,
        org=org,
        repo=repo,
        tag=tag,
        prefix=prefix,
    )