Gitea Client

HTTP client for the Gitea package registry API.

GiteaClient handles pagination, SSL configuration, authentication, and file downloads. PackageFile is the data class returned by iter_package_files().

devpi_gitea_sync.gitea

GiteaError

Bases: RuntimeError

Raised when an unexpected response is received from Gitea.

Source code in devpi_gitea_sync/gitea.py

class GiteaError(RuntimeError):
    """Raised when an unexpected response is received from Gitea."""

PackageFile dataclass

Represents a single downloadable file exposed by the Gitea packages API.

Parameters:

Name	Type	Description	Default
id	int	The ID of the package file.	required
name	str	The name of the package file.	required
download_url	str	The URL to download the package file.	required
size	int	The size of the package file in bytes.	required
package	str	The name of the package.	required
version	str	The version of the package.	required
repository	str | None	The name of the repository the package belongs to.	required

Source code in devpi_gitea_sync/gitea.py

@dataclass
class PackageFile:
    """Represents a single downloadable file exposed by the Gitea packages API.

    :param id: The ID of the package file.
    :param name: The name of the package file.
    :param download_url: The URL to download the package file.
    :param size: The size of the package file in bytes.
    :param package: The name of the package.
    :param version: The version of the package.
    :param repository: The name of the repository the package belongs to.
    """

    id: int
    name: str
    download_url: str
    size: int
    package: str
    version: str
    repository: str | None

    def looks_like_distribution(self) -> bool:
        """Return True when the asset name resembles a Python distribution archive."""
        lowered = self.name.lower()
        return any(lowered.endswith(suffix) for suffix in PYTHON_DIST_SUFFIXES)

looks_like_distribution()

Return True when the asset name resembles a Python distribution archive.

Source code in devpi_gitea_sync/gitea.py

def looks_like_distribution(self) -> bool:
    """Return True when the asset name resembles a Python distribution archive."""
    lowered = self.name.lower()
    return any(lowered.endswith(suffix) for suffix in PYTHON_DIST_SUFFIXES)

GiteaClient

Minimal wrapper around the bits of the Gitea API we need.

Source code in devpi_gitea_sync/gitea.py

class GiteaClient:
    """Minimal wrapper around the bits of the Gitea API we need."""

    def __init__(self, config: GiteaConfig, token: str):
        """Initialise the HTTP session that will speak to the Gitea API.

        :param config: The Gitea configuration.
        :param token: The Gitea API token.
        :raises ValueError: If the token is not provided.
        """
        if not token:
            raise ValueError("GiteaClient requires a personal access token.")
        self._base_url = config.url.rstrip("/")
        headers = {
            "User-Agent": "devpi-gitea-sync/0.1.0",
            "Authorization": f"token {token}",
        }
        self._timeout = config.timeout
        self._session = requests.Session()
        self._session.verify = config.verify_ssl
        self._session.headers.update(headers)

        if not config.verify_ssl:
            disable_warnings(exceptions.InsecureRequestWarning)
            logger.debug(
                "SSL verification disabled for Gitea client targeting %s; suppressing warnings.",
                self._base_url,
            )

    def close(self) -> None:
        """Close the underlying HTTP session."""
        self._session.close()

    def __enter__(self) -> "GiteaClient":
        """Support use as a context manager."""
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        """Ensure resources are cleaned up when leaving a context."""
        self.close()

    def list_packages(self, organization: str, *, package_type: str = "pypi") -> list[dict]:
        """Return metadata for packages published under the organisation.

        :param organization: The name of the Gitea organization.
        :param package_type: The type of packages to list.
        :return: A list of package metadata dictionaries.
        :raises GiteaError: If the API request fails.
        """
        logger.debug(
            "Fetching packages for organization '%s' (type=%s)", organization, package_type
        )
        packages: list[dict] = []
        page = 1
        limit = 50
        while True:
            params = {"limit": limit, "page": page}
            if package_type:
                params["type"] = package_type
            payload = self._request_json(
                f"/api/v1/packages/{organization}", params=params, allow_404=True
            )
            if payload is None:
                logger.info(
                    "Packages endpoint returned 404 for organization '%s'; no packages available.",
                    organization,
                )
                break
            if not isinstance(payload, list):
                raise GiteaError(
                    f"Expected list response for packages in organization '{organization}', "
                    f"got {type(payload).__name__}."
                )
            if not payload:
                break
            packages.extend(payload)
            if len(payload) < limit:
                break
            page += 1
        if not packages:
            logger.info(
                "Gitea returned no packages for organization '%s' (type=%s).",
                organization,
                package_type,
            )
        else:
            logger.debug(
                "Retrieved %d packages for organization '%s' (type=%s).",
                len(packages),
                organization,
                package_type,
            )
        return packages

    def iter_package_files(
        self, organization: str, *, package_type: str = "pypi"
    ) -> Iterator[PackageFile]:
        """Yield downloadable package files for the organisation.

        :param organization: The name of the Gitea organization.
        :param package_type: The type of packages to list.
        :return: An iterator of PackageFile objects.
        """
        packages = self.list_packages(organization, package_type=package_type)
        if not packages:
            return
        logger.debug(
            "Iterating package files for %d packages in organization '%s'.",
            len(packages),
            organization,
        )
        for package_entry in packages:
            if package_type and _extract_package_type(package_entry) != package_type:
                continue

            package_name = _extract_package_name(package_entry)
            if not package_name:
                continue

            repository = _extract_repository_name(package_entry)
            version_hint = _extract_version_string(package_entry)

            if version_hint:
                version_entries = [package_entry]
            else:
                version_entries = self._list_package_versions(
                    organization, package_name, package_type=package_type
                )

            for version_entry in version_entries:
                version, files = self._extract_version_files(
                    organization,
                    package_type,
                    package_name,
                    version_entry,
                )
                if not version or not files:
                    continue
                for file_entry in files:
                    asset = self._build_package_file(
                        organization=organization,
                        package_type=package_type,
                        package_name=package_name,
                        version=version,
                        file_entry=file_entry,
                        repository=repository,
                    )
                    if asset:
                        yield asset

    def download_asset(
        self,
        asset: PackageFile,
        *,
        chunk_size: int = 1 << 14,
        dest_dir: Path | None = None,
    ) -> Path:
        """Download a package file to a local directory and return its path.

        :param asset: The PackageFile to download.
        :param chunk_size: The chunk size for streaming the download.
        :param dest_dir: The destination directory; uses the system temp dir when not set.
        :return: The path to the downloaded file.
        :raises GiteaError: If the download fails.
        """
        try:
            response = self._session.get(asset.download_url, stream=True, timeout=self._timeout)
            response.raise_for_status()
        except requests.RequestException as exc:
            raise GiteaError(
                f"Failed to download asset '{asset.name}' "
                f"from repository '{asset.repository}': {exc}"
            ) from exc

        filename = _resolve_filename(asset.name, response.headers.get("Content-Disposition"))
        dest_base = Path(tempfile.gettempdir()) if dest_dir is None else dest_dir
        dest_base.mkdir(parents=True, exist_ok=True)
        target_path = dest_base / filename

        with tempfile.NamedTemporaryFile(delete=False, dir=str(dest_base)) as fp:
            for chunk in response.iter_content(chunk_size=chunk_size):
                if not chunk:
                    continue
                fp.write(chunk)
            temp_path = Path(fp.name)

        temp_path.replace(target_path)

        return target_path

    def _request_json(
        self,
        path: str,
        *,
        params: dict | None = None,
        allow_404: bool = False,
    ) -> object:
        """Perform a GET request returning JSON content.

        :param path: The API path to request.
        :param params: A dictionary of query parameters.
        :param allow_404: If True, a 404 response will be treated as an empty result.
        :return: The JSON response as a Python object.
        :raises GiteaError: If the request fails.
        """
        url = f"{self._base_url}{path}"
        try:
            response = self._session.get(url, params=params, timeout=self._timeout)
            if allow_404 and response.status_code == 404:
                logger.debug("Request to %s returned 404; treating as empty result.", path)
                return None
            response.raise_for_status()
        except requests.RequestException as exc:
            raise GiteaError(f"Request to {path} failed: {exc}") from exc
        return response.json()

    def _list_package_versions(
        self,
        organization: str,
        package_name: str,
        *,
        package_type: str,
    ) -> Sequence[dict]:
        """Return version metadata entries for a package.

        :param organization: The name of the Gitea organization.
        :param package_name: The name of the package.
        :param package_type: The type of the package.
        :return: A sequence of version metadata dictionaries.
        :raises GiteaError: If the API request fails.
        """
        encoded_name = quote(package_name, safe="")
        path = f"/api/v1/packages/{organization}/{package_type}/{encoded_name}"
        payload = self._request_json(path, allow_404=True)
        if payload is None:
            logger.debug(
                "No versions endpoint found for package '%s' in organization '%s'.",
                package_name,
                organization,
            )
            return []
        if isinstance(payload, list):
            logger.debug(
                "Found %d versions for package '%s' in organization '%s'.",
                len(payload),
                package_name,
                organization,
            )
            return payload
        if isinstance(payload, dict):
            versions = payload.get("versions") or payload.get("Versions")
            if isinstance(versions, list):
                logger.debug(
                    "Found %d versions for package '%s' in organization '%s'.",
                    len(versions),
                    package_name,
                    organization,
                )
                return versions
        raise GiteaError(
            f"Unexpected response while listing versions for package '{package_name}'."
        )

    def _extract_version_files(
        self,
        organization: str,
        package_type: str,
        package_name: str,
        version_entry: dict,
    ) -> tuple[str | None, Sequence[dict]]:
        """Return version string and file list for a version entry.

        :param organization: The name of the Gitea organization.
        :param package_type: The type of the package.
        :param package_name: The name of the package.
        :param version_entry: The version metadata dictionary.
        :return: A tuple of the version string and a sequence of file metadata dictionaries.
        """
        version = (
            version_entry.get("version")
            or version_entry.get("name")
            or version_entry.get("Version")
            or version_entry.get("Name")
        )
        if not version:
            return None, []
        files = version_entry.get("files") or version_entry.get("Files")
        if not files:
            files = self._list_version_files(
                organization=organization,
                package_name=package_name,
                package_type=package_type,
                version=version,
            )
        return version, files or []

    def _list_version_files(
        self,
        *,
        organization: str,
        package_name: str,
        package_type: str,
        version: str,
    ) -> Sequence[dict]:
        """Retrieve all files published for a given package version.

        :param organization: The name of the Gitea organization.
        :param package_name: The name of the package.
        :param package_type: The type of the package.
        :param version: The version of the package.
        :return: A sequence of file metadata dictionaries.
        :raises GiteaError: If the API request fails.
        """
        encoded_name = quote(package_name, safe="")
        encoded_version = quote(version, safe="")
        path = (
            f"/api/v1/packages/{organization}/{package_type}/"
            f"{encoded_name}/{encoded_version}/files"
        )
        payload = self._request_json(path, allow_404=True)

        if payload is None:
            logger.debug(
                "No files found for package %s/%s@%s (type=%s).",
                organization,
                package_name,
                version,
                package_type,
            )
            return []

        if isinstance(payload, list):
            logger.debug(
                "Retrieved %d files for package %s/%s@%s (type=%s).",
                len(payload),
                organization,
                package_name,
                version,
                package_type,
            )
            return payload

        if isinstance(payload, dict):
            files = payload.get("files") or payload.get("Files") or []
            logger.debug(
                "Retrieved %d files for package %s/%s@%s (type=%s).",
                len(files),
                organization,
                package_name,
                version,
                package_type,
            )
            return files

        raise GiteaError(
            f"Unexpected response for package files {organization}/{package_name}@{version}."
        )

    def _build_package_file(
        self,
        *,
        organization: str,
        package_type: str,
        package_name: str,
        version: str,
        file_entry: dict,
        repository: str | None,
    ) -> PackageFile | None:
        """Create a PackageFile instance from raw API metadata.

        :param organization: The name of the Gitea organization.
        :param package_type: The type of the package.
        :param package_name: The name of the package.
        :param version: The version of the package.
        :param file_entry: The file metadata dictionary.
        :param repository: The name of the repository the package belongs to.
        :return: A PackageFile instance, or None if the file entry is invalid.
        """
        name = file_entry.get("name") or file_entry.get("Name")
        if not name:
            return None
        download_url = (
            file_entry.get("download_url")
            or file_entry.get("DownloadURL")
            or file_entry.get("browser_download_url")
        )
        if not download_url:
            encoded_package = quote(package_name, safe="")
            encoded_version = quote(version, safe="")
            encoded_filename = quote(name, safe="")
            download_url = (
                f"{self._base_url}/api/packages/{organization}/{package_type}/"
                f"files/{encoded_package}/{encoded_version}/{encoded_filename}"
            )
        size = file_entry.get("size") or file_entry.get("Size") or 0
        identifier = file_entry.get("id") or file_entry.get("ID") or 0
        return PackageFile(
            id=int(identifier),
            name=name,
            download_url=download_url,
            size=int(size),
            package=package_name,
            version=version,
            repository=repository,
        )

close()

Close the underlying HTTP session.

Source code in devpi_gitea_sync/gitea.py

def close(self) -> None:
    """Close the underlying HTTP session."""
    self._session.close()

list_packages(organization, *, package_type='pypi')

Return metadata for packages published under the organisation.

Parameters:

Name	Type	Description	Default
organization	str	The name of the Gitea organization.	required
package_type	str	The type of packages to list.	'pypi'

Returns:

Type	Description
list[dict]	A list of package metadata dictionaries.

Raises:

Type	Description
GiteaError	If the API request fails.

Source code in devpi_gitea_sync/gitea.py

def list_packages(self, organization: str, *, package_type: str = "pypi") -> list[dict]:
    """Return metadata for packages published under the organisation.

    :param organization: The name of the Gitea organization.
    :param package_type: The type of packages to list.
    :return: A list of package metadata dictionaries.
    :raises GiteaError: If the API request fails.
    """
    logger.debug(
        "Fetching packages for organization '%s' (type=%s)", organization, package_type
    )
    packages: list[dict] = []
    page = 1
    limit = 50
    while True:
        params = {"limit": limit, "page": page}
        if package_type:
            params["type"] = package_type
        payload = self._request_json(
            f"/api/v1/packages/{organization}", params=params, allow_404=True
        )
        if payload is None:
            logger.info(
                "Packages endpoint returned 404 for organization '%s'; no packages available.",
                organization,
            )
            break
        if not isinstance(payload, list):
            raise GiteaError(
                f"Expected list response for packages in organization '{organization}', "
                f"got {type(payload).__name__}."
            )
        if not payload:
            break
        packages.extend(payload)
        if len(payload) < limit:
            break
        page += 1
    if not packages:
        logger.info(
            "Gitea returned no packages for organization '%s' (type=%s).",
            organization,
            package_type,
        )
    else:
        logger.debug(
            "Retrieved %d packages for organization '%s' (type=%s).",
            len(packages),
            organization,
            package_type,
        )
    return packages

iter_package_files(organization, *, package_type='pypi')

Yield downloadable package files for the organisation.

Parameters:

Name	Type	Description	Default
organization	str	The name of the Gitea organization.	required
package_type	str	The type of packages to list.	'pypi'

Returns:

Type	Description
Iterator[PackageFile]	An iterator of PackageFile objects.

Source code in devpi_gitea_sync/gitea.py

def iter_package_files(
    self, organization: str, *, package_type: str = "pypi"
) -> Iterator[PackageFile]:
    """Yield downloadable package files for the organisation.

    :param organization: The name of the Gitea organization.
    :param package_type: The type of packages to list.
    :return: An iterator of PackageFile objects.
    """
    packages = self.list_packages(organization, package_type=package_type)
    if not packages:
        return
    logger.debug(
        "Iterating package files for %d packages in organization '%s'.",
        len(packages),
        organization,
    )
    for package_entry in packages:
        if package_type and _extract_package_type(package_entry) != package_type:
            continue

        package_name = _extract_package_name(package_entry)
        if not package_name:
            continue

        repository = _extract_repository_name(package_entry)
        version_hint = _extract_version_string(package_entry)

        if version_hint:
            version_entries = [package_entry]
        else:
            version_entries = self._list_package_versions(
                organization, package_name, package_type=package_type
            )

        for version_entry in version_entries:
            version, files = self._extract_version_files(
                organization,
                package_type,
                package_name,
                version_entry,
            )
            if not version or not files:
                continue
            for file_entry in files:
                asset = self._build_package_file(
                    organization=organization,
                    package_type=package_type,
                    package_name=package_name,
                    version=version,
                    file_entry=file_entry,
                    repository=repository,
                )
                if asset:
                    yield asset

download_asset(asset, *, chunk_size=1 << 14, dest_dir=None)

Download a package file to a local directory and return its path.

Parameters:

Name	Type	Description	Default
asset	PackageFile	The PackageFile to download.	required
chunk_size	int	The chunk size for streaming the download.	1 << 14
dest_dir	Path | None	The destination directory; uses the system temp dir when not set.	None

Returns:

Type	Description
Path	The path to the downloaded file.

Raises:

Type	Description
GiteaError	If the download fails.

Source code in devpi_gitea_sync/gitea.py

def download_asset(
    self,
    asset: PackageFile,
    *,
    chunk_size: int = 1 << 14,
    dest_dir: Path | None = None,
) -> Path:
    """Download a package file to a local directory and return its path.

    :param asset: The PackageFile to download.
    :param chunk_size: The chunk size for streaming the download.
    :param dest_dir: The destination directory; uses the system temp dir when not set.
    :return: The path to the downloaded file.
    :raises GiteaError: If the download fails.
    """
    try:
        response = self._session.get(asset.download_url, stream=True, timeout=self._timeout)
        response.raise_for_status()
    except requests.RequestException as exc:
        raise GiteaError(
            f"Failed to download asset '{asset.name}' "
            f"from repository '{asset.repository}': {exc}"
        ) from exc

    filename = _resolve_filename(asset.name, response.headers.get("Content-Disposition"))
    dest_base = Path(tempfile.gettempdir()) if dest_dir is None else dest_dir
    dest_base.mkdir(parents=True, exist_ok=True)
    target_path = dest_base / filename

    with tempfile.NamedTemporaryFile(delete=False, dir=str(dest_base)) as fp:
        for chunk in response.iter_content(chunk_size=chunk_size):
            if not chunk:
                continue
            fp.write(chunk)
        temp_path = Path(fp.name)

    temp_path.replace(target_path)

    return target_path