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