Skip to content

Commit

Permalink
storeapi: rename SnapClientIndex to SnapAPI
Browse files Browse the repository at this point in the history
This is the client side API on api.snapcraft.io. Rename and adapt
docstrings to align with reality.

Signed-off-by: Sergio Schvezov <sergio.schvezov@canonical.com>
  • Loading branch information
sergiusens committed Feb 24, 2021
1 parent cee67ae commit 8de1efe
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 23 deletions.
Expand Up @@ -15,22 +15,27 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.

import contextlib
import logging
import os
from typing import Dict

from . import _macaroon_auth, constants, errors, logger
from . import _macaroon_auth, constants, errors
from ._client import Client
from .info import SnapInfo


class SnapIndexClient(Client):
"""The Click Package Index knows everything about existing snaps.
https://wiki.ubuntu.com/AppStore/Interfaces/ClickPackageIndex is the
canonical reference.
logger = logging.getLogger(__name__)


class SnapAPI(Client):
"""The Snap API is used to query snaps.
This is an interface to query that API which is documented
at http://api.snapcraft.io/docs/.
"""

def __init__(self, conf):
"""Initialize the SnapIndexClient object.
"""Initialize SnapAPI.
:param config conf: Configuration details for the client.
:type config: snapcraft.config.Config
Expand All @@ -39,8 +44,9 @@ def __init__(self, conf):
conf, os.environ.get("STORE_API_URL", constants.STORE_API_URL),
)

def get_default_headers(self, api="v2"):
def _get_default_headers(self, api="v2"):
"""Return default headers for CPI requests.
Tries to build an 'Authorization' header with local credentials
if they are available.
Also pin specific branded store if `SNAPCRAFT_UBUNTU_STORE`
Expand All @@ -51,12 +57,12 @@ def get_default_headers(self, api="v2"):
with contextlib.suppress(errors.InvalidCredentialsError):
headers["Authorization"] = _macaroon_auth(self.conf)

branded_store = os.getenv("SNAPCRAFT_UBUNTU_STORE")
if branded_store:
brand_store = os.getenv("SNAPCRAFT_UBUNTU_STORE")
if brand_store:
if api == "v2":
headers["Snap-Device-Store"] = branded_store
headers["Snap-Device-Store"] = brand_store
elif api == "v1":
headers["X-Ubuntu-Store"] = branded_store
headers["X-Ubuntu-Store"] = brand_store
else:
logger.warning("Incorrect API version passed: {!r}.".format(api))

Expand All @@ -68,10 +74,10 @@ def get_info(self, snap_name: str, *, arch: str = None) -> SnapInfo:
:param str snap_name: Name of the snap.
:param str arch: Architecture of the snap (none by default).
:return information for the snap.
:returns: information for the snap.
:rtype: SnapInfo
"""
headers = self.get_default_headers()
headers = self._get_default_headers()
headers.update(
{
"Accept": "application/json",
Expand All @@ -97,14 +103,14 @@ def get_info(self, snap_name: str, *, arch: str = None) -> SnapInfo:
def get_assertion(
self, assertion_type: str, snap_id: str
) -> Dict[str, Dict[str, str]]:
"""Get the assertion for the specified snap.
"""Get the assertion of assertion_type for the specified snap_id.
:param str assertion_type: The type of the assertion.
:param str assertion_type: The type of assertion.
:param str snap_id: The ID of the snap.
:return Assertion for the snap.
:returns: Assertion for the snap.
"""
headers = self.get_default_headers(api="v1")
headers = self._get_default_headers(api="v1")
logger.debug("Getting snap-declaration for {}".format(snap_id))
url = "/api/v1/snaps/assertions/{}/{}/{}".format(
assertion_type, constants.DEFAULT_SERIES, snap_id
Expand Down
12 changes: 6 additions & 6 deletions snapcraft/storeapi/_store_client.py
Expand Up @@ -27,7 +27,7 @@

from . import _upload, errors, logger
from ._sca_client import SCAClient
from ._snap_index_client import SnapIndexClient
from ._snap_api import SnapAPI
from ._snap_v2_client import SnapV2Client
from ._sso_client import SSOClient
from ._up_down_client import UpDownClient
Expand All @@ -36,13 +36,13 @@


class StoreClient:
"""High-level client for the V2.0 API SCA resources."""
"""High-level client Snap resources."""

def __init__(self) -> None:
super().__init__()
self.conf = config.Config()
self.sso = SSOClient(self.conf)
self.cpi = SnapIndexClient(self.conf)
self.snap = SnapAPI(self.conf)
self.updown = UpDownClient(self.conf)
self.sca = SCAClient(self.conf)
self.v2_snap = SnapV2Client(self.conf)
Expand Down Expand Up @@ -143,7 +143,7 @@ def acl(self) -> Dict[str, Any]:
return acl_data

def get_snap_name_for_id(self, snap_id: str) -> str:
declaration_assertion = self.cpi.get_assertion("snap-declaration", snap_id)
declaration_assertion = self.snap.get_assertion("snap-declaration", snap_id)
return declaration_assertion["headers"]["snap-name"]

def verify_acl(self) -> Dict[str, Union[List[str], str]]:
Expand Down Expand Up @@ -261,7 +261,7 @@ def download(
arch: Optional[str] = None,
except_hash: str = ""
):
snap_info = self.cpi.get_info(snap_name)
snap_info = self.snap.get_info(snap_name)
channel_mapping = snap_info.get_channel_mapping(
risk=risk, track=track, arch=arch
)
Expand Down Expand Up @@ -297,7 +297,7 @@ def _download_snap(self, download_details, download_path):
if resume_possible and os.path.exists(download_path):
total_read = os.path.getsize(download_path)
headers["Range"] = "bytes={}-".format(total_read)
request = self.cpi.get(download_url, headers=headers, stream=True)
request = self.snap.get(download_url, headers=headers, stream=True)
request.raise_for_status()
redirections = [h.headers["Location"] for h in request.history]
if redirections:
Expand Down

0 comments on commit 8de1efe

Please sign in to comment.