Skip to content

Class swarmauri_storage_github_release.gh_release_storage_adapter.GithubReleaseStorageAdapter

swarmauri_storage_github_release.gh_release_storage_adapter.GithubReleaseStorageAdapter 

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

Bases: StorageAdapterBase

Storage adapter that uses GitHub Releases to store and retrieve assets.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
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
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

Return the base ghrel:// URI for this adapter.

upload 

upload(key, data)

Upload data under key as a release asset and return the artifact URI.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def upload(self, key: str, data: BinaryIO) -> str:
    """Upload ``data`` under ``key`` as a release asset and return the artifact URI."""
    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)

Return the bytes of asset key as a BytesIO object.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def download(self, key: str) -> BinaryIO:
    """Return the bytes of asset ``key`` as a BytesIO object."""
    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='')

Upload all files under src using an optional prefix.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
123
124
125
126
127
128
129
130
131
def upload_dir(self, src: str | os.PathLike, *, prefix: str = "") -> None:
    """Upload all files under *src* using an optional *prefix*."""
    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)

Yield asset keys under prefix.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
133
134
135
136
137
138
139
140
141
142
def iter_prefix(self, prefix: str):
    """Yield asset keys under *prefix*."""
    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_dir 

download_dir(prefix, dest_dir)

Download all assets under prefix into dest_dir.

Source code in swarmauri_storage_github_release/gh_release_storage_adapter.py 
144
145
146
147
148
149
150
151
152
def download_dir(self, prefix: str, dest_dir: str | os.PathLike) -> None:
    """Download all assets under *prefix* into *dest_dir*."""
    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_storage_github_release/gh_release_storage_adapter.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
@classmethod
def from_uri(cls, uri: str) -> "GithubReleaseStorageAdapter":
    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("adapters", {}).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,
    )