Module materialize.bintray

A Bintray API client.

Consult the official Bintray API documentation for details. Only the operations that are presently necessary are implemented.

Expand source code Browse git
# Copyright Materialize, Inc. All rights reserved.
#
# Use of this software is governed by the Business Source License
# included in the LICENSE file at the root of this repository.
#
# As of the Change Date specified in that file, in accordance with
# the Business Source License, use of this software will be governed
# by the Apache License, Version 2.0.

"""A Bintray API client.

Consult the [official Bintray API documentation][official-docs] for details.
Only the operations that are presently necessary are implemented.

[official-docs]: https://bintray.com/docs/api/
"""

from typing import IO, Sequence
from typing_extensions import Final
from requests import Session, Response
import requests

API_BASE: Final = "https://api.bintray.com"


class VersionAlreadyExistsError(Exception):
    """Raised when creating a version of a Bintray package that already
    exists."""


class PackageClient:
    """A client for a specific Bintray package."""

    def __init__(self, session: Session, subject: str, repo: str, package: str) -> None:
        self.session = session
        self.subject = subject
        self.repo = repo
        self.package = package

    def create_version(self, version: str, desc: str, vcs_tag: str) -> Response:
        """Create a new version of the package.

        Args:
            version: The name of the version to create, e.g. "2.0".
            desc: A description of the version.
            vcs_tag: The version control tag or commit corresponding to the
                version.
        """
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions"
        res = self.session.post(
            url, json={"name": version, "desc": desc, "vcs_tag": vcs_tag}
        )
        if res.status_code == requests.codes.conflict:
            raise VersionAlreadyExistsError()
        else:
            res.raise_for_status()
        return res

    def delete_version(self, version: str) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
        res = self.session.delete(url)
        res.raise_for_status()
        return res

    def debian_upload(
        self,
        version: str,
        path: str,
        data: IO[bytes],
        distributions: Sequence[str],
        components: Sequence[str],
        architectures: Sequence[str],
    ) -> Response:
        """Upload a Debian artifact for a version.

        The version must already have been created with e.g.
        `PackageClient.create_version`.

        Args:
            version: The version of the package to attach the artifact to.
            path: The file path for the uploaded .deb.
            data: An IO handle to the .deb file.
            distributions: The Debian distributions for the artifact.
            components: The components in the Debian distribution to which the
                artifact belongs.
            architecture: The architectures for which the artifact was built.
        """
        url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/{path}"
        res = self.session.put(
            url,
            data=data,
            headers={
                "X-Bintray-Debian-Distribution": ",".join(distributions),
                "X-Bintray-Debian-Component": ",".join(components),
                "X-Bintray-Debian-Architecture": ",".join(architectures),
            },
        )
        res.raise_for_status()
        return res

    def publish_uploads(self, version: str, wait_for_seconds: int = 0) -> Response:
        """Publish all unpublished content for a version.

        Args:
            version: The version of the package to publish content for.
            wait_for_seconds: The number of seconds to wait for files to
                publish. A value of -1 means to wait for the maximum allowable
                time. A value of 0 is the default and means to publish
                asynchronously, without waiting at all.
        """
        url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/publish"
        res = self.session.post(url, json={"publish_wait_for_secs": wait_for_seconds})
        res.raise_for_status()
        return res

    def get_metadata(self) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}"
        res = self.session.get(url)
        return res

    def get_version(self, version: str) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
        res = self.session.get(url)
        res.raise_for_status()
        return res


class RepoClient:
    """A client for a specific Bintray repository."""

    def __init__(self, session: Session, subject: str, repo: str):
        self.session = session
        self.subject = subject
        self.repo = repo

    def package(self, package: str) -> PackageClient:
        """Construct a client for the named package."""
        return PackageClient(self.session, self.subject, self.repo, package)


class Client:
    """A root Bintray client.

    Args:
        subject: The organization to issue requests for.
        user: The user to authenticate as.
        api_key: The API key to authenticate with.
    """

    def __init__(self, subject: str, user: str, api_key: str):
        self.session = Session()
        self.session.auth = (f"{user}", api_key)
        self.subject = subject

    def repo(self, repo: str) -> RepoClient:
        """Construct a client for the named repository."""
        return RepoClient(self.session, self.subject, repo)

Classes

class Client (subject: str, user: str, api_key: str)

A root Bintray client.

Args

subject
The organization to issue requests for.
user
The user to authenticate as.
api_key
The API key to authenticate with.
Expand source code Browse git
class Client:
    """A root Bintray client.

    Args:
        subject: The organization to issue requests for.
        user: The user to authenticate as.
        api_key: The API key to authenticate with.
    """

    def __init__(self, subject: str, user: str, api_key: str):
        self.session = Session()
        self.session.auth = (f"{user}", api_key)
        self.subject = subject

    def repo(self, repo: str) -> RepoClient:
        """Construct a client for the named repository."""
        return RepoClient(self.session, self.subject, repo)

Methods

def repo(self, repo: str) -> RepoClient

Construct a client for the named repository.

Expand source code Browse git
def repo(self, repo: str) -> RepoClient:
    """Construct a client for the named repository."""
    return RepoClient(self.session, self.subject, repo)
class PackageClient (session: requests.sessions.Session, subject: str, repo: str, package: str)

A client for a specific Bintray package.

Expand source code Browse git
class PackageClient:
    """A client for a specific Bintray package."""

    def __init__(self, session: Session, subject: str, repo: str, package: str) -> None:
        self.session = session
        self.subject = subject
        self.repo = repo
        self.package = package

    def create_version(self, version: str, desc: str, vcs_tag: str) -> Response:
        """Create a new version of the package.

        Args:
            version: The name of the version to create, e.g. "2.0".
            desc: A description of the version.
            vcs_tag: The version control tag or commit corresponding to the
                version.
        """
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions"
        res = self.session.post(
            url, json={"name": version, "desc": desc, "vcs_tag": vcs_tag}
        )
        if res.status_code == requests.codes.conflict:
            raise VersionAlreadyExistsError()
        else:
            res.raise_for_status()
        return res

    def delete_version(self, version: str) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
        res = self.session.delete(url)
        res.raise_for_status()
        return res

    def debian_upload(
        self,
        version: str,
        path: str,
        data: IO[bytes],
        distributions: Sequence[str],
        components: Sequence[str],
        architectures: Sequence[str],
    ) -> Response:
        """Upload a Debian artifact for a version.

        The version must already have been created with e.g.
        `PackageClient.create_version`.

        Args:
            version: The version of the package to attach the artifact to.
            path: The file path for the uploaded .deb.
            data: An IO handle to the .deb file.
            distributions: The Debian distributions for the artifact.
            components: The components in the Debian distribution to which the
                artifact belongs.
            architecture: The architectures for which the artifact was built.
        """
        url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/{path}"
        res = self.session.put(
            url,
            data=data,
            headers={
                "X-Bintray-Debian-Distribution": ",".join(distributions),
                "X-Bintray-Debian-Component": ",".join(components),
                "X-Bintray-Debian-Architecture": ",".join(architectures),
            },
        )
        res.raise_for_status()
        return res

    def publish_uploads(self, version: str, wait_for_seconds: int = 0) -> Response:
        """Publish all unpublished content for a version.

        Args:
            version: The version of the package to publish content for.
            wait_for_seconds: The number of seconds to wait for files to
                publish. A value of -1 means to wait for the maximum allowable
                time. A value of 0 is the default and means to publish
                asynchronously, without waiting at all.
        """
        url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/publish"
        res = self.session.post(url, json={"publish_wait_for_secs": wait_for_seconds})
        res.raise_for_status()
        return res

    def get_metadata(self) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}"
        res = self.session.get(url)
        return res

    def get_version(self, version: str) -> Response:
        url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
        res = self.session.get(url)
        res.raise_for_status()
        return res

Methods

def create_version(self, version: str, desc: str, vcs_tag: str) -> requests.models.Response

Create a new version of the package.

Args

version
The name of the version to create, e.g. "2.0".
desc
A description of the version.
vcs_tag
The version control tag or commit corresponding to the version.
Expand source code Browse git
def create_version(self, version: str, desc: str, vcs_tag: str) -> Response:
    """Create a new version of the package.

    Args:
        version: The name of the version to create, e.g. "2.0".
        desc: A description of the version.
        vcs_tag: The version control tag or commit corresponding to the
            version.
    """
    url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions"
    res = self.session.post(
        url, json={"name": version, "desc": desc, "vcs_tag": vcs_tag}
    )
    if res.status_code == requests.codes.conflict:
        raise VersionAlreadyExistsError()
    else:
        res.raise_for_status()
    return res
def debian_upload(self, version: str, path: str, data: IO[bytes], distributions: Sequence[str], components: Sequence[str], architectures: Sequence[str]) -> requests.models.Response

Upload a Debian artifact for a version.

The version must already have been created with e.g. PackageClient.create_version().

Args

version
The version of the package to attach the artifact to.
path
The file path for the uploaded .deb.
data
An IO handle to the .deb file.
distributions
The Debian distributions for the artifact.
components
The components in the Debian distribution to which the artifact belongs.
architecture
The architectures for which the artifact was built.
Expand source code Browse git
def debian_upload(
    self,
    version: str,
    path: str,
    data: IO[bytes],
    distributions: Sequence[str],
    components: Sequence[str],
    architectures: Sequence[str],
) -> Response:
    """Upload a Debian artifact for a version.

    The version must already have been created with e.g.
    `PackageClient.create_version`.

    Args:
        version: The version of the package to attach the artifact to.
        path: The file path for the uploaded .deb.
        data: An IO handle to the .deb file.
        distributions: The Debian distributions for the artifact.
        components: The components in the Debian distribution to which the
            artifact belongs.
        architecture: The architectures for which the artifact was built.
    """
    url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/{path}"
    res = self.session.put(
        url,
        data=data,
        headers={
            "X-Bintray-Debian-Distribution": ",".join(distributions),
            "X-Bintray-Debian-Component": ",".join(components),
            "X-Bintray-Debian-Architecture": ",".join(architectures),
        },
    )
    res.raise_for_status()
    return res
def delete_version(self, version: str) -> requests.models.Response
Expand source code Browse git
def delete_version(self, version: str) -> Response:
    url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
    res = self.session.delete(url)
    res.raise_for_status()
    return res
def get_metadata(self) -> requests.models.Response
Expand source code Browse git
def get_metadata(self) -> Response:
    url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}"
    res = self.session.get(url)
    return res
def get_version(self, version: str) -> requests.models.Response
Expand source code Browse git
def get_version(self, version: str) -> Response:
    url = f"{API_BASE}/packages/{self.subject}/{self.repo}/{self.package}/versions/{version}"
    res = self.session.get(url)
    res.raise_for_status()
    return res
def publish_uploads(self, version: str, wait_for_seconds: int = 0) -> requests.models.Response

Publish all unpublished content for a version.

Args

version
The version of the package to publish content for.
wait_for_seconds
The number of seconds to wait for files to publish. A value of -1 means to wait for the maximum allowable time. A value of 0 is the default and means to publish asynchronously, without waiting at all.
Expand source code Browse git
def publish_uploads(self, version: str, wait_for_seconds: int = 0) -> Response:
    """Publish all unpublished content for a version.

    Args:
        version: The version of the package to publish content for.
        wait_for_seconds: The number of seconds to wait for files to
            publish. A value of -1 means to wait for the maximum allowable
            time. A value of 0 is the default and means to publish
            asynchronously, without waiting at all.
    """
    url = f"{API_BASE}/content/{self.subject}/{self.repo}/{self.package}/{version}/publish"
    res = self.session.post(url, json={"publish_wait_for_secs": wait_for_seconds})
    res.raise_for_status()
    return res
class RepoClient (session: requests.sessions.Session, subject: str, repo: str)

A client for a specific Bintray repository.

Expand source code Browse git
class RepoClient:
    """A client for a specific Bintray repository."""

    def __init__(self, session: Session, subject: str, repo: str):
        self.session = session
        self.subject = subject
        self.repo = repo

    def package(self, package: str) -> PackageClient:
        """Construct a client for the named package."""
        return PackageClient(self.session, self.subject, self.repo, package)

Methods

def package(self, package: str) -> PackageClient

Construct a client for the named package.

Expand source code Browse git
def package(self, package: str) -> PackageClient:
    """Construct a client for the named package."""
    return PackageClient(self.session, self.subject, self.repo, package)
class VersionAlreadyExistsError (...)

Raised when creating a version of a Bintray package that already exists.

Expand source code Browse git
class VersionAlreadyExistsError(Exception):
    """Raised when creating a version of a Bintray package that already
    exists."""

Ancestors

  • builtins.Exception
  • builtins.BaseException