Skip to content

funfedi_results.codeberg

CodebergPackageDownloader dataclass

Enables downloading the latest version of a package

Parameters:

Name Type Description Default
package PackageDescription
required
owner str
'funfedidev'
api_prefix str
'https://codeberg.org/api'
Source code in funfedi_results/codeberg/__init__.py
@dataclass
class CodebergPackageDownloader:
    """Enables downloading the latest version of a package"""

    package: PackageDescription
    owner: str = "funfedidev"

    api_prefix: str = "https://codeberg.org/api"

    @property
    def filelist_url(self):
        """
        ```
        >>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
        >>> downloader.filelist_url
        'https://codeberg.org/api/v1/packages/funfedidev/generic/results_funfedi_connect/0.1.3/files'

        ```
        """
        return f"{self.api_prefix}/v1/packages/{self.owner}/generic/{self.package.path}/files"

    def download_link(self, app_name, app_version):
        """
        ```
        >>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
        >>> downloader.download_link("cattle_grid", "0.5.20")
        'https://codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.3/cattle_grid_0.5.20.zip'

        ```
        """
        filename = f"{app_name}_{app_version}.zip"
        return f"{self.api_prefix}/packages/{self.owner}/generic/{self.package.path}/{filename}"

    def has_a_version(self, app_name, app_version):
        filename = f"{app_name}_{app_version}.zip"
        return filename in self.fetch_filelist()

    def fetch_filelist(self) -> list[str]:
        result = requests.get(self.filelist_url)

        if result.status_code == 404:
            return []

        result.raise_for_status()
        data = result.json()
        return [x.get("name") for x in data]

    def latest_version_per_app(self):
        """Returns the latest versions"""
        names = self.fetch_filelist()
        return names_to_latest_dict(names)

    def target_directory(self, directory: Path):
        return directory / self.package.path

    def download_latest(self, directory: Path):
        names_and_versions = self.latest_version_per_app()
        target_directory = self.target_directory(directory)
        target_directory.mkdir(exist_ok=True, parents=True)

        for name, version in names_and_versions.items():
            download_link = self.download_link(name, version)
            filename = target_directory / download_link.split("/")[-1]

            if filename.exists():
                continue

            print(f"downloading {download_link}")

            response = requests.get(download_link)
            assert response.status_code == 200, (
                f"Got response code {response.status_code} for downloading {download_link}"
            )
            with open(filename, "wb") as f:
                f.write(response.content)

filelist_url property

filelist_url
>>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
>>> downloader.filelist_url
'https://codeberg.org/api/v1/packages/funfedidev/generic/results_funfedi_connect/0.1.3/files'
download_link(app_name, app_version)
>>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
>>> downloader.download_link("cattle_grid", "0.5.20")
'https://codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.3/cattle_grid_0.5.20.zip'
Source code in funfedi_results/codeberg/__init__.py
def download_link(self, app_name, app_version):
    """
    ```
    >>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
    >>> downloader.download_link("cattle_grid", "0.5.20")
    'https://codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.3/cattle_grid_0.5.20.zip'

    ```
    """
    filename = f"{app_name}_{app_version}.zip"
    return f"{self.api_prefix}/packages/{self.owner}/generic/{self.package.path}/{filename}"

latest_version_per_app

latest_version_per_app()

Returns the latest versions

Source code in funfedi_results/codeberg/__init__.py
def latest_version_per_app(self):
    """Returns the latest versions"""
    names = self.fetch_filelist()
    return names_to_latest_dict(names)

CodebergUploader dataclass

Helps uploading files to codeberg

>>> uploader = CodebergUploader("user", "token")
>>> uploader.upload_url(PackageDescription(KnownPackages.connect, "0.1.2"), "app_v1.2.3.zip")
'https://token:user@codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.2/app_v1.2.3.zip'

Parameters:

Name Type Description Default
token str
required
user str
required
Source code in funfedi_results/codeberg/uploader.py
@dataclass
class CodebergUploader:
    """
    Helps uploading files to codeberg

    ```
    >>> uploader = CodebergUploader("user", "token")
    >>> uploader.upload_url(PackageDescription(KnownPackages.connect, "0.1.2"), "app_v1.2.3.zip")
    'https://token:user@codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.2/app_v1.2.3.zip'

    ```
    """

    token: str
    user: str

    @property
    def api_prefix(self) -> str:
        return f"https://{self.user}:{self.token}@codeberg.org/api/packages/funfedidev/generic/"

    def upload_url(self, package: PackageDescription, filename):
        return self.api_prefix + f"{package.path}/{filename}"

    def upload_file(self, package: PackageDescription, filename, file):
        """Uploads a file with url as in upload_url"""

        files = {"file": open(file, "rb")}
        url = self.upload_url(package, filename)
        result = requests.put(url, files=files)
        if result.status_code != 201:
            print(f"Upload to codeberg failed with status {result.status_code}")
            print(result.text)
        else:
            print("Uploaded to codeberg")

    @staticmethod
    def from_env():
        user = os.environ.get("CODEBERG_USER")
        token = os.environ.get("CODEBERG_TOKEN")

        if user is None or token is None:
            raise Exception("Expected CODEBERG_USER and CODEBERG_TOKEN to be set")

        return CodebergUploader(user, token)

upload_file

upload_file(package: PackageDescription, filename, file)

Uploads a file with url as in upload_url

Source code in funfedi_results/codeberg/uploader.py
def upload_file(self, package: PackageDescription, filename, file):
    """Uploads a file with url as in upload_url"""

    files = {"file": open(file, "rb")}
    url = self.upload_url(package, filename)
    result = requests.put(url, files=files)
    if result.status_code != 201:
        print(f"Upload to codeberg failed with status {result.status_code}")
        print(result.text)
    else:
        print("Uploaded to codeberg")

KnownPackages

Bases: StrEnum

Predefined packages

Source code in funfedi_results/codeberg/package_description.py
class KnownPackages(StrEnum):
    """Predefined packages"""

    parsing = "results_funfedi_parsing"
    connect = "results_funfedi_connect"

PackageDescription dataclass

Describes a package

Parameters:

Name Type Description Default
name str | KnownPackages
required
version str
required
Source code in funfedi_results/codeberg/package_description.py
@dataclass
class PackageDescription:
    """Describes a package"""

    name: str | KnownPackages
    version: str

    @property
    def path(self):
        return f"{self.name}/{self.version}"

    def __post_init__(self):
        if to_semver(self.version) == semver.Version(0):
            raise Exception("version should be a valid version")