vscode marketplace api list all extensions + download artifacts

vscode_market_place_like_api_download_extensions_automatically.py

"""
Inspired by

https://open-vsx.org/swagger-ui/index.html?urls.primaryName=VSCode%20Adapter#/vs-code-api/extensionQuery
https://gist.github.com/jossef/8d7681ac0c7fd28e93147aa5044bc129
"""

import itertools
from dataclasses import dataclass
from typing import Any, Optional

import pyrfc6266
import requests
from requests.adapters import HTTPAdapter, Retry


@dataclass
class Info:
    user: str
    name: str
    target: str
    version: str
    deps: list[str]

    @property
    def extension_id(self):
        return f"{self.user}.{self.name}"

    @property
    def download_url(self):
        url = (
            f"publishers/{self.user}/vsextensions/{self.name}/{self.version}/vspackage"
        )
        if self.target is not None:
            url += f"?targetPlatform={self.target}"
        return url


def __select_highest_priority_value(values, priority, transform=lambda x: x):
    # Create a priority dictionary from the priority list
    priority_dict = {value: index for index, value in enumerate(priority)}

    # Initialize variables to keep track of the highest priority value and its priority index
    highest_priority_value = None
    highest_priority_index = len(priority)

    # Iterate through the values and find the one with the highest priority
    for value in values:
        transformed_value = transform(value)
        if (
            transformed_value in priority_dict
            and priority_dict[transformed_value] < highest_priority_index
        ):
            highest_priority_value = value
            highest_priority_index = priority_dict[transformed_value]

    return highest_priority_value


def __get_vscode_extensions(
    url: str,
    extensions: list[str],
    max_page=10000,
    page_size=100,
    include_versions=True,
    include_files=True,
    include_category_and_tags=True,
    include_shared_accounts=True,
    include_version_properties=True,
    exclude_non_validated=True,
    include_installation_targets=True,
    include_asset_uri=True,
    include_statistics=False,
    include_latest_version_only=True,
    unpublished=False,
    include_name_conflict_info=True,
    api_version="7.2-preview.1",
    session=None,
):
    if not session:
        session = requests.session()

    headers = {"Accept": f"application/json; charset=utf-8; api-version={api_version}"}

    flags = 0
    if include_versions:
        flags |= 0x1

    if include_files:
        flags |= 0x2

    if include_category_and_tags:
        flags |= 0x4

    if include_shared_accounts:
        flags |= 0x8

    if include_shared_accounts:
        flags |= 0x8

    if include_version_properties:
        flags |= 0x10

    if exclude_non_validated:
        flags |= 0x20

    if include_installation_targets:
        flags |= 0x40

    if include_asset_uri:
        flags |= 0x80

    if include_statistics:
        flags |= 0x100

    if include_latest_version_only:
        flags |= 0x200

    if unpublished:
        flags |= 0x1000

    if include_name_conflict_info:
        flags |= 0x8000

    for page in range(1, max_page + 1):
        body = {
            "filters": [
                {
                    "criteria": [
                        {"filterType": 8, "value": "Microsoft.VisualStudio.Code"},
                        {
                            "filterType": 10,
                            "value": ",".join(extensions),
                        },
                    ],
                    "pageNumber": page,
                    "pageSize": page_size,
                    "sortBy": 0,
                    "sortOrder": 0,
                }
            ],
            "assetTypes": [],
            "flags": flags,
        }

        r = session.post(
            f"{url}/extensionquery",
            json=body,
            headers=headers,
        )
        r.raise_for_status()
        response = r.json()
        yield from response["results"][0]["extensions"]

        if len(extensions) != page_size:
            break


def __process_extension_api_output(
    extension: dict[str, Any], targets: list[Optional[str]]
):
    version = __select_highest_priority_value(
        extension["versions"], targets, lambda version: version.get("targetPlatform")
    )
    props = version["properties"]
    target = version["targetPlatform"] if "targetPlatform" in version else None
    return Info(
        user=extension["publisher"]["publisherName"],
        name=extension["extensionName"],
        version=version["version"],
        target=target,
        deps=[
            dep
            for dep in itertools.chain.from_iterable(
                (
                    kv["value"].split(",")
                    for kv in props
                    if kv["key"] == "Microsoft.VisualStudio.Code.ExtensionPack"
                    or kv["key"] == "Microsoft.VisualStudio.Code.ExtensionDependencies"
                )
            )
            if len(dep) != 0
        ],
    )


def __download_url(session, url):
    """
    Download a file from a specified URL and save it locally.

    Args:
        session: The requests session to use for the download.
        url (str): The URL of the file to download.
    """
    response = session.get(url, stream=True)
    response.raise_for_status()  # Raise an HTTPError for bad responses (4xx and 5xx)
    local_filename = pyrfc6266.requests_response_to_filename(response)

    with open(local_filename, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)

    print(f"File downloaded and saved as {local_filename}")


def __main(url: str, extensions: list[str], targets: list[Optional[str]]):
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "OPTIONS"],
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session = requests.Session()
    session.mount("https://", adapter)
    session.mount("http://", adapter)

    for extension_output in __get_vscode_extensions(
        url=url, extensions=extensions, session=session
    ):
        extension = __process_extension_api_output(
            extension=extension_output, targets=targets
        )
        if extension.extension_id not in extensions:
            continue
        extensions.extend(extension.deps)
    extensions = list(set(extensions))
    print(extensions)
    for extension_output in __get_vscode_extensions(
        url=url, extensions=extensions, session=session
    ):
        extension = __process_extension_api_output(
            extension=extension_output, targets=targets
        )
        if extension.extension_id not in extensions:
            continue
        print(extension)
        print(extension.download_url)
        print(__download_url(url=f"{url}/{extension.download_url}", session=session))


if __name__ == "__main__":
    __main(
        url="https://open-vsx.org/vscode/gallery",
        extensions=["streetsidesoftware.code-spell-checker"],
        targets=["linux-x64", "web", "universal", None],
    )