Skip to content

funfedi_results.codeberg

BadCodebergCredentialsException

Bases: Exception

Thrown if CodebergUploader.from_env doesn’t find the credentials

Source code in funfedi_results/codeberg/uploader.py 
 9
10
class BadCodebergCredentialsException(Exception):
    """Thrown if CodebergUploader.from_env doesn't find the credentials"""

CodebergPackageDownloader dataclass

Enables downloading the latest version of a package

Parameters:

Name Type Description Default
package PackageDescription
required
owner str
'funfedidev'
base_url str
'https://codeberg.org/'
api_prefix str
'https://codeberg.org/api'
Source code in funfedi_results/codeberg/__init__.py 
 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
103
104
105
106
107
108
109
110
111
112
113
114
115
@dataclass
class CodebergPackageDownloader:
    """Enables downloading the latest version of a package"""

    package: PackageDescription
    owner: str = "funfedidev"

    base_url: str = "https://codeberg.org/"
    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"

    @property
    def download_url(self) -> str:
        """
        Returns to the url of the codeberg download page

        ```
        >>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
        >>> downloader.download_url
        'https://codeberg.org/funfedidev/-/packages/generic/results_funfedi_connect'

        ```
        """
        return f"{self.base_url}{self.owner}/-/packages/generic/{self.package.name}"

    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) -> bool:
        """Returns True if a result file already exists for app_name and 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):
        """Downloads the latest zip files and deposits them in directory"""

        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

            logger.info(f"downloading {download_link}")

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

download_url property 

download_url: str

Returns to the url of the codeberg download page

>>> downloader = CodebergPackageDownloader(PackageDescription("results_funfedi_connect", "0.1.3"))
>>> downloader.download_url
'https://codeberg.org/funfedidev/-/packages/generic/results_funfedi_connect'

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_latest 

download_latest(directory: Path)

Downloads the latest zip files and deposits them in directory

Source code in funfedi_results/codeberg/__init__.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def download_latest(self, directory: Path):
    """Downloads the latest zip files and deposits them in directory"""

    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

        logger.info(f"downloading {download_link}")

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

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 
58
59
60
61
62
63
64
65
66
67
68
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}"

has_a_version 

has_a_version(app_name, app_version) -> bool

Returns True if a result file already exists for app_name and app_version

Source code in funfedi_results/codeberg/__init__.py 
70
71
72
73
def has_a_version(self, app_name, app_version) -> bool:
    """Returns True if a result file already exists for app_name and app_version"""
    filename = f"{app_name}_{app_version}.zip"
    return filename in self.fetch_filelist()

latest_version_per_app 

latest_version_per_app()

Returns the latest versions

Source code in funfedi_results/codeberg/__init__.py 
85
86
87
88
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://user:token@codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.2/app_v1.2.3.zip'

Parameters:

Name Type Description Default
user str
required
token str
required
Source code in funfedi_results/codeberg/uploader.py 
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
60
61
62
63
64
65
66
67
68
69
70
71
72
@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://user:token@codeberg.org/api/packages/funfedidev/generic/results_funfedi_connect/0.1.2/app_v1.2.3.zip'

    ```
    """

    user: str
    token: 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")

    def delete_file(self, package: PackageDescription, filename: str):
        """Deletes a file"""
        url = self.upload_url(package, filename)
        result = requests.delete(url)

        if result.status_code == 204:
            print("Delete successful")
        else:
            print("Delete failed")
            print(result)
            print(result.text)
            sys.exit(1)

    @staticmethod
    def from_env():
        """Initializes uploader from CODERBERG_USER and CODEBERG_TOKEN environment variables"""
        user = os.environ.get("CODEBERG_USER")
        token = os.environ.get("CODEBERG_TOKEN")

        if not user or not token:
            raise BadCodebergCredentialsException(
                "Expected CODEBERG_USER and CODEBERG_TOKEN to be set"
            )

        return CodebergUploader(user=user, token=token)

delete_file 

delete_file(package: PackageDescription, filename: str)

Deletes a file

Source code in funfedi_results/codeberg/uploader.py 
48
49
50
51
52
53
54
55
56
57
58
59
def delete_file(self, package: PackageDescription, filename: str):
    """Deletes a file"""
    url = self.upload_url(package, filename)
    result = requests.delete(url)

    if result.status_code == 204:
        print("Delete successful")
    else:
        print("Delete failed")
        print(result)
        print(result.text)
        sys.exit(1)

from_env staticmethod 

from_env()

Initializes uploader from CODERBERG_USER and CODEBERG_TOKEN environment variables

Source code in funfedi_results/codeberg/uploader.py 
61
62
63
64
65
66
67
68
69
70
71
72
@staticmethod
def from_env():
    """Initializes uploader from CODERBERG_USER and CODEBERG_TOKEN environment variables"""
    user = os.environ.get("CODEBERG_USER")
    token = os.environ.get("CODEBERG_TOKEN")

    if not user or not token:
        raise BadCodebergCredentialsException(
            "Expected CODEBERG_USER and CODEBERG_TOKEN to be set"
        )

    return CodebergUploader(user=user, token=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 
36
37
38
39
40
41
42
43
44
45
46
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 
 8
 9
10
11
12
13
14
class KnownPackages(StrEnum):
    """Predefined packages"""

    parsing = "results_funfedi_parsing"
    parsing_notes = "results_funfedi_parsing_notes"
    parsing_events = "results_funfedi_parsing_events"
    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 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@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")