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 DebAlreadyExistsError(Exception):
"""Raised when uploading a Debian 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),
},
)
if res.status_code == requests.codes.conflict:
raise DebAlreadyExistsError()
else:
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 DebAlreadyExistsError (...)
-
Raised when uploading a Debian package that already exists.
Expand source code Browse git
class DebAlreadyExistsError(Exception): """Raised when uploading a Debian package that already exists."""
Ancestors
- builtins.Exception
- builtins.BaseException
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), }, ) if res.status_code == requests.codes.conflict: raise DebAlreadyExistsError() else: 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), }, ) if res.status_code == requests.codes.conflict: raise DebAlreadyExistsError() else: 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