plain.vendor

  1import re
  2
  3import requests
  4import tomlkit
  5
  6from plain.assets.finders import APP_ASSETS_DIR
  7
  8from .exceptions import (
  9    UnknownContentTypeError,
 10    VersionMismatchError,
 11)
 12
 13VENDOR_DIR = APP_ASSETS_DIR / "vendor"
 14
 15
 16def iter_next_version(version):
 17    if len(version.split(".")) == 2:
 18        major, minor = version.split(".")
 19        yield f"{int(major) + 1}.0"
 20        yield f"{major}.{int(minor) + 1}"
 21    elif len(version.split(".")) == 3:
 22        major, minor, patch = version.split(".")
 23        yield f"{int(major) + 1}.0.0"
 24        yield f"{major}.{int(minor) + 1}.0"
 25        yield f"{major}.{minor}.{int(patch) + 1}"
 26    else:
 27        raise ValueError(f"Unable to iterate next version for {version}")
 28
 29
 30class Dependency:
 31    def __init__(self, name, **config):
 32        self.name = name
 33        self.url = config.get("url", "")
 34        self.installed = config.get("installed", "")
 35        self.filename = config.get("filename", "")
 36        self.sourcemap = config.get("sourcemap", "")
 37
 38    @staticmethod
 39    def parse_version_from_url(url):
 40        if match := re.search(r"\d+\.\d+\.\d+", url):
 41            return match.group(0)
 42
 43        if match := re.search(r"\d+\.\d+", url):
 44            return match.group(0)
 45
 46        return ""
 47
 48    def __str__(self):
 49        return f"{self.name} -> {self.url}"
 50
 51    def download(self, version):
 52        # If the string contains a {version} placeholder, replace it
 53        download_url = self.url.replace("{version}", version)
 54
 55        response = requests.get(download_url)
 56        response.raise_for_status()
 57
 58        content_type = response.headers.get("content-type")
 59        if content_type.lower() not in (
 60            "application/javascript; charset=utf-8",
 61            "text/javascript; charset=utf-8",
 62            "application/json; charset=utf-8",
 63            "text/css; charset=utf-8",
 64        ):
 65            raise UnknownContentTypeError(
 66                f"Unknown content type for {self.name}: {content_type}"
 67            )
 68
 69        # Good chance it will redirect to a more final URL (which we hope is versioned)
 70        url = response.url
 71        version = self.parse_version_from_url(url)
 72
 73        return version, response
 74
 75    def install(self):
 76        if self.installed:
 77            version, response = self.download(self.installed)
 78            if version != self.installed:
 79                raise VersionMismatchError(
 80                    f"Version mismatch for {self.name}: {self.installed} != {version}"
 81                )
 82            return self.vendor(response)
 83        else:
 84            return self.update()
 85
 86    def update(self):
 87        def try_version(v):
 88            try:
 89                version, response = self.download(v)
 90                return version, response
 91            except requests.RequestException:
 92                return "", None
 93
 94        if not self.installed:
 95            # If we don't know the installed version yet,
 96            # just use the url as given
 97            version, response = self.download("")
 98        else:
 99            version, response = try_version("latest")  # A lot of CDNs support this
100            if not version:
101                # Try the next few versions
102                for v in iter_next_version(self.installed):
103                    version, response = try_version(v)
104                    if version:
105                        break
106
107                    # TODO ideally this would keep going -- if we move to 2.0, and no 3.0, try 2.1, 2.2, etc.
108
109        if not version:
110            # Use the currently installed version if we found nothing else
111            version, response = self.download(self.installed)
112
113        vendored_path = self.vendor(response)
114        self.installed = version
115
116        if self.installed:
117            # If the exact version was in the string, replace it with {version} placeholder
118            self.url = self.url.replace(self.installed, "{version}")
119
120        self.save_config()
121        return vendored_path
122
123    def save_config(self):
124        with open("pyproject.toml") as f:
125            pyproject = tomlkit.load(f)
126
127        # Force [tool.plain.vendor.dependencies] to be a table
128        dependencies = tomlkit.table()
129        dependencies.update(
130            pyproject.get("tool", {})
131            .get("plain", {})
132            .get("vendor", {})
133            .get("dependencies", {})
134        )
135
136        # Force [tool.plain.vendor.dependencies.{name}] to be an inline table
137        # name = { url = "https://example.com", installed = "1.0.0" }
138        dependencies[self.name] = tomlkit.inline_table()
139        dependencies[self.name]["url"] = self.url
140        dependencies[self.name]["installed"] = self.installed
141        if self.filename:
142            dependencies[self.name]["filename"] = self.filename
143        if self.sourcemap:
144            dependencies[self.name]["sourcemap"] = self.sourcemap
145
146        # Have to give it the right structure in case they don't exist
147        if "tool" not in pyproject:
148            pyproject["tool"] = tomlkit.table()
149        if "plain" not in pyproject["tool"]:
150            pyproject["tool"]["plain"] = tomlkit.table()
151        if "vendor" not in pyproject["tool"]["plain"]:
152            pyproject["tool"]["plain"]["vendor"] = tomlkit.table()
153
154        pyproject["tool"]["plain"]["vendor"]["dependencies"] = dependencies
155
156        with open("pyproject.toml", "w") as f:
157            f.write(tomlkit.dumps(pyproject))
158
159    def vendor(self, response):
160        if not VENDOR_DIR.exists():
161            VENDOR_DIR.mkdir(parents=True)
162
163        if self.filename:
164            # Use a specific filename from config
165            filename = self.filename
166        else:
167            # Otherwise, use the filename from the URL
168            filename = response.url.split("/")[-1]
169
170        vendored_path = VENDOR_DIR / filename
171
172        with open(vendored_path, "wb") as f:
173            f.write(response.content)
174
175        # If a sourcemap is requested, download it as well
176        if self.sourcemap:
177            if isinstance(self.sourcemap, str):
178                # Use a specific filename from config
179                sourcemap_filename = self.sourcemap
180            else:
181                # Otherwise, append .map to the URL
182                sourcemap_filename = f"{filename}.map"
183
184            sourcemap_url = "/".join(
185                response.url.split("/")[:-1] + [sourcemap_filename]
186            )
187            sourcemap_response = requests.get(sourcemap_url)
188            sourcemap_response.raise_for_status()
189
190            sourcemap_path = VENDOR_DIR / sourcemap_filename
191
192            with open(sourcemap_path, "wb") as f:
193                f.write(sourcemap_response.content)
194
195        return vendored_path
196
197
198def get_deps():
199    with open("pyproject.toml") as f:
200        pyproject = tomlkit.load(f)
201
202    config = (
203        pyproject.get("tool", {})
204        .get("plain", {})
205        .get("vendor", {})
206        .get("dependencies", {})
207    )
208
209    deps = []
210
211    for name, data in config.items():
212        deps.append(Dependency(name, **data))
213
214    return deps