Coverage for flogin/pip.py: 100%

1from __future__ import annotations

3import logging

4import subprocess

5import sys

6import tempfile

7from pathlib import Path

9from .utils import MISSING

11try:

12 import requests

13except ImportError: # cov: skip

14 requests = MISSING

16from .errors import PipExecutionError, UnableToDownloadPip

18TYPE_CHECKING = False

19if TYPE_CHECKING:

20 from types import (

21 TracebackType, # https://github.com/astral-sh/ruff/issues/15681

22 )

23 from typing import Self

25__all__ = ("Pip",)

27log = logging.getLogger(__name__)

30class Pip:

31 r"""This is a helper class for dealing with pip in a production environment.

33 When flow launcher installs python, it does not install pip. Because of that, this class will temp-install pip while you need it, then delete it when you're done with it.

35 .. versionadded:: 2.0.0

37 .. WARNING::

38 This class is blocking, and should only be used before you load your plugin.

40 Parameters

41 -----------

42 libs_dir: Optional[:class:`pathlib.Path` | :class:`str`]

43 The directory that your plugin's dependencies are installed to. Defaults to ``lib``.

45 Example

46 -------

47 This should be used before your plugin gets loaded. Here is an example of what your main file might look like when using pip:

49 .. code-block:: py3

51 # Add your paths

52 sys.path.append(...)

53 sys.path.append(...)

55 from flogin import Pip

57 with Pip() as pip:

58 pip.ensure_installed("msgspec") # ensure the msgspec package is installed correctly

60 # import and run your plugin

61 from plugin.plugin import YourPlugin

62 YourPlugin().run()

64 .. container::

66 .. describe:: with Pip(...) as pip:

68 :class:`Pip` can be used as a context manager, with :meth:`Pip.download_pip` being called on enter and :meth:`Pip.delete_pip` being called on exit.

69 """

71 _libs_dir: Path

73 def __init__(self, libs_dir: Path | str = "lib") -> None:

74 if requests is MISSING:

75 raise ImportError(

76 "Pip's Extra Dependencies are not installed. You can install them with flogin[pip]"

77 )

79 self._pip_fp: Path | None = None

80 self.libs_dir = libs_dir

82 @property

83 def libs_dir(self) -> Path:

84 """:class:`pathlib.Path`: The directory that your plugin's dependencies are installed to."""

85 return self._libs_dir

87 @libs_dir.setter

88 def libs_dir(self, new: Path | str) -> None:

89 if isinstance(new, str):

90 new = Path(new)

92 if not new.exists():

93 # Despite the fact that installing works perfectly fine with a nonexistent directory

94 # adding it to path before its created then trying to import from it doesn't.

95 raise ValueError(f"Directory Not Found: {new}")

97 self._libs_dir = new

99 def download_pip(self) -> None:

100 r"""Downloads the temp version of pip from pypa.

101

102 .. NOTE::

103 This is automatically called when using :class:`Pip` as a context manager.

104

105 Raises

106 ------

107 :class:`UnableToDownloadPip`

108 This is raised when an error occured while attempting to download pip.

109

110 Returns

111 -------

112 ``None``

113 """

114 try:

115 res = requests.get("https://bootstrap.pypa.io/pip/pip.pyz", timeout=10)

116 res.raise_for_status()

117 except requests.RequestException as error:

118 raise UnableToDownloadPip(error) from error

119

120 error = None

121 with tempfile.NamedTemporaryFile("wb", suffix="-pip.pyz", delete=False) as f:

122 try:

123 f.write(res.content)

124 self._pip_fp = Path(f.name)

125 except BaseException as e:

126 error = e

127

128 if error:

129 Path(f.name).unlink(missing_ok=True)

130 raise error

131

132 def delete_pip(self) -> None:

133 r"""Deletes the temp version of pip installed on the system.

134

135 .. NOTE::

136 This is automatically called when using :class:`Pip` as a context manager.

137

138 Returns

139 --------

140 ``None``

141 """

142

143 if self._pip_fp:

144 self._pip_fp.unlink(missing_ok=True)

145 log.info("Pip deleted from %s", self._pip_fp)

146

147 def __enter__(self) -> Self:

148 self.download_pip()

149 return self

150

151 def __exit__(

152 self,

153 type_: type[BaseException] | None,

154 value: BaseException | None,

155 traceback: TracebackType | None,

156 ) -> bool:

157 self.delete_pip()

158 return False

159

160 def run(self, *args: str) -> str:

161 r"""Runs a pip CLI command.

162

163 This method is used to interact directly with pip.

164

165 .. NOTE::

166 This method can not be used until :meth:`download_pip` is ran, which you can do by calling it manually or using :class:`Pip` as a context manager.

167

168 Parameters

169 -----------

170 \*args: :class:`str`

171 The args that should be passed to pip. Ex: ``help``.

172

173 Raises

174 ------

175 :class:`~flogin.errors.PipExecutionError`

176 This is raised when the returncode that pip gives indicates an error.

177 :class:`RuntimeError`

178 This is raised when :meth:`Pip.download_pip` has not ran yet.

179

180 Returns

181 --------

182 :class:`str`

183 The output from pip.

184 """

185

186 if self._pip_fp is None:

187 raise RuntimeError("Pip has not been installed")

188

189 pip = self._pip_fp.as_posix()

190 cmd = [sys.executable, pip, *args]

191 log.debug("Sending command: %r", cmd)

192

193 try:

194 proc = subprocess.run(cmd, capture_output=True, check=True)

195 except subprocess.CalledProcessError as e:

196 log.debug("Pip command failed. stdout: %r, stderr: %r", e.output, e.stderr)

197 raise PipExecutionError(e)

198

199 output = proc.stdout.decode()

200 log.debug("Pip stdout: %r", output)

201 if proc.stderr:

202 log.debug("Pip stderr: %r", proc.stderr)

203

204 return output

205

206 def install_packages(self, *packages: str) -> None:

207 r"""An easy way to install packages for your plugin.

208

209 .. NOTE::

210 The packages will be installed to the directory set in :attr:`Pip.libs_dir`.

211

212 Parameters

213 ----------

214 \*packages: :class:`str`

215 The name of the packages on PyPi that you want to install.

216

217 Raises

218 ------

219 :class:`PipException`

220 This is raised when the returncode that pip gives indicates an error.

221

222 Returns

223 -------

224 ``None``

225 """

226

227 self.run(

228 "install",

229 "--upgrade",

230 "--force-reinstall",

231 *packages,

232 "-t",

233 self.libs_dir.as_posix(),

234 )

235

236 def ensure_installed(self, package: str, *, module: str | None = None) -> bool:

237 r"""Ensures a package is properly installed, and if not, reinstalls it.

238

239 Parameters

240 ----------

241 package: :class:`str`

242 The name of the package on PyPi that you want to install.

243 module: Optional[:class:`str`]

244 The name of the module you want to check to see if its installed. Defaults to the ``package`` value.

245

246 Raises

247 ------

248 :class:`PipException`

249 This is raised when the returncode that pip gives indicates an error.

250

251 Returns

252 -------

253 :class:`bool`

254 ``True`` indicates that the package wasn't properly installed, and was successfully reinstalled.

255 ``False`` indicates that the package was already properly installed.

256 """

257

258 try:

259 __import__(module or package)

260 except (ImportError, ModuleNotFoundError):

261 self.install_packages(package)

262 return True

263 return False

264

265 def freeze(self) -> list[str]:

266 r"""Returns a list of installed packages from ``pip freeze``.

267

268 .. NOTE::

269 The directory checked for packages is set in :attr:`Pip.libs_dir`.

270

271 Raises

272 ------

273 :class:`PipException`

274 This is raised when the returncode that pip gives indicates an error.

275

276 Returns

277 --------

278 list[:class:`str`]

279 The list of packages and versions.

280 """

281

282 return self.run("freeze", "--path", self.libs_dir.as_posix()).splitlines()