main.py 12 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382
  1. import io
  2. import logging
  3. import os
  4. import shutil
  5. import sys
  6. import tempfile
  7. from collections import OrderedDict
  8. from contextlib import contextmanager
  9. from typing import (IO, Dict, Iterable, Iterator, Mapping, Optional, Tuple,
  10. Union)
  11. from .parser import Binding, parse_stream
  12. from .variables import parse_variables
  13. # A type alias for a string path to be used for the paths in this file.
  14. # These paths may flow to `open()` and `shutil.move()`; `shutil.move()`
  15. # only accepts string paths, not byte paths or file descriptors. See
  16. # https://github.com/python/typeshed/pull/6832.
  17. StrPath = Union[str, 'os.PathLike[str]']
  18. logger = logging.getLogger(__name__)
  19. def with_warn_for_invalid_lines(mappings: Iterator[Binding]) -> Iterator[Binding]:
  20. for mapping in mappings:
  21. if mapping.error:
  22. logger.warning(
  23. "Python-dotenv could not parse statement starting at line %s",
  24. mapping.original.line,
  25. )
  26. yield mapping
  27. class DotEnv:
  28. def __init__(
  29. self,
  30. dotenv_path: Optional[StrPath],
  31. stream: Optional[IO[str]] = None,
  32. verbose: bool = False,
  33. encoding: Optional[str] = None,
  34. interpolate: bool = True,
  35. override: bool = True,
  36. ) -> None:
  37. self.dotenv_path: Optional[StrPath] = dotenv_path
  38. self.stream: Optional[IO[str]] = stream
  39. self._dict: Optional[Dict[str, Optional[str]]] = None
  40. self.verbose: bool = verbose
  41. self.encoding: Optional[str] = encoding
  42. self.interpolate: bool = interpolate
  43. self.override: bool = override
  44. @contextmanager
  45. def _get_stream(self) -> Iterator[IO[str]]:
  46. if self.dotenv_path and os.path.isfile(self.dotenv_path):
  47. with open(self.dotenv_path, encoding=self.encoding) as stream:
  48. yield stream
  49. elif self.stream is not None:
  50. yield self.stream
  51. else:
  52. if self.verbose:
  53. logger.info(
  54. "Python-dotenv could not find configuration file %s.",
  55. self.dotenv_path or '.env',
  56. )
  57. yield io.StringIO('')
  58. def dict(self) -> Dict[str, Optional[str]]:
  59. """Return dotenv as dict"""
  60. if self._dict:
  61. return self._dict
  62. raw_values = self.parse()
  63. if self.interpolate:
  64. self._dict = OrderedDict(resolve_variables(raw_values, override=self.override))
  65. else:
  66. self._dict = OrderedDict(raw_values)
  67. return self._dict
  68. def parse(self) -> Iterator[Tuple[str, Optional[str]]]:
  69. with self._get_stream() as stream:
  70. for mapping in with_warn_for_invalid_lines(parse_stream(stream)):
  71. if mapping.key is not None:
  72. yield mapping.key, mapping.value
  73. def set_as_environment_variables(self) -> bool:
  74. """
  75. Load the current dotenv as system environment variable.
  76. """
  77. if not self.dict():
  78. return False
  79. for k, v in self.dict().items():
  80. if k in os.environ and not self.override:
  81. continue
  82. if v is not None:
  83. os.environ[k] = v
  84. return True
  85. def get(self, key: str) -> Optional[str]:
  86. """
  87. """
  88. data = self.dict()
  89. if key in data:
  90. return data[key]
  91. if self.verbose:
  92. logger.warning("Key %s not found in %s.", key, self.dotenv_path)
  93. return None
  94. def get_key(
  95. dotenv_path: StrPath,
  96. key_to_get: str,
  97. encoding: Optional[str] = "utf-8",
  98. ) -> Optional[str]:
  99. """
  100. Get the value of a given key from the given .env.
  101. Returns `None` if the key isn't found or doesn't have a value.
  102. """
  103. return DotEnv(dotenv_path, verbose=True, encoding=encoding).get(key_to_get)
  104. @contextmanager
  105. def rewrite(
  106. path: StrPath,
  107. encoding: Optional[str],
  108. ) -> Iterator[Tuple[IO[str], IO[str]]]:
  109. if not os.path.isfile(path):
  110. with open(path, mode="w", encoding=encoding) as source:
  111. source.write("")
  112. with tempfile.NamedTemporaryFile(mode="w", encoding=encoding, delete=False) as dest:
  113. try:
  114. with open(path, encoding=encoding) as source:
  115. yield (source, dest)
  116. except BaseException:
  117. os.unlink(dest.name)
  118. raise
  119. shutil.move(dest.name, path)
  120. def set_key(
  121. dotenv_path: StrPath,
  122. key_to_set: str,
  123. value_to_set: str,
  124. quote_mode: str = "always",
  125. export: bool = False,
  126. encoding: Optional[str] = "utf-8",
  127. ) -> Tuple[Optional[bool], str, str]:
  128. """
  129. Adds or Updates a key/value to the given .env
  130. If the .env path given doesn't exist, fails instead of risking creating
  131. an orphan .env somewhere in the filesystem
  132. """
  133. if quote_mode not in ("always", "auto", "never"):
  134. raise ValueError(f"Unknown quote_mode: {quote_mode}")
  135. quote = (
  136. quote_mode == "always"
  137. or (quote_mode == "auto" and not value_to_set.isalnum())
  138. )
  139. if quote:
  140. value_out = "'{}'".format(value_to_set.replace("'", "\\'"))
  141. else:
  142. value_out = value_to_set
  143. if export:
  144. line_out = f'export {key_to_set}={value_out}\n'
  145. else:
  146. line_out = f"{key_to_set}={value_out}\n"
  147. with rewrite(dotenv_path, encoding=encoding) as (source, dest):
  148. replaced = False
  149. missing_newline = False
  150. for mapping in with_warn_for_invalid_lines(parse_stream(source)):
  151. if mapping.key == key_to_set:
  152. dest.write(line_out)
  153. replaced = True
  154. else:
  155. dest.write(mapping.original.string)
  156. missing_newline = not mapping.original.string.endswith("\n")
  157. if not replaced:
  158. if missing_newline:
  159. dest.write("\n")
  160. dest.write(line_out)
  161. return True, key_to_set, value_to_set
  162. def unset_key(
  163. dotenv_path: StrPath,
  164. key_to_unset: str,
  165. quote_mode: str = "always",
  166. encoding: Optional[str] = "utf-8",
  167. ) -> Tuple[Optional[bool], str]:
  168. """
  169. Removes a given key from the given `.env` file.
  170. If the .env path given doesn't exist, fails.
  171. If the given key doesn't exist in the .env, fails.
  172. """
  173. if not os.path.exists(dotenv_path):
  174. logger.warning("Can't delete from %s - it doesn't exist.", dotenv_path)
  175. return None, key_to_unset
  176. removed = False
  177. with rewrite(dotenv_path, encoding=encoding) as (source, dest):
  178. for mapping in with_warn_for_invalid_lines(parse_stream(source)):
  179. if mapping.key == key_to_unset:
  180. removed = True
  181. else:
  182. dest.write(mapping.original.string)
  183. if not removed:
  184. logger.warning("Key %s not removed from %s - key doesn't exist.", key_to_unset, dotenv_path)
  185. return None, key_to_unset
  186. return removed, key_to_unset
  187. def resolve_variables(
  188. values: Iterable[Tuple[str, Optional[str]]],
  189. override: bool,
  190. ) -> Mapping[str, Optional[str]]:
  191. new_values: Dict[str, Optional[str]] = {}
  192. for (name, value) in values:
  193. if value is None:
  194. result = None
  195. else:
  196. atoms = parse_variables(value)
  197. env: Dict[str, Optional[str]] = {}
  198. if override:
  199. env.update(os.environ) # type: ignore
  200. env.update(new_values)
  201. else:
  202. env.update(new_values)
  203. env.update(os.environ) # type: ignore
  204. result = "".join(atom.resolve(env) for atom in atoms)
  205. new_values[name] = result
  206. return new_values
  207. def _walk_to_root(path: str) -> Iterator[str]:
  208. """
  209. Yield directories starting from the given directory up to the root
  210. """
  211. if not os.path.exists(path):
  212. raise IOError('Starting path not found')
  213. if os.path.isfile(path):
  214. path = os.path.dirname(path)
  215. last_dir = None
  216. current_dir = os.path.abspath(path)
  217. while last_dir != current_dir:
  218. yield current_dir
  219. parent_dir = os.path.abspath(os.path.join(current_dir, os.path.pardir))
  220. last_dir, current_dir = current_dir, parent_dir
  221. def find_dotenv(
  222. filename: str = '.env',
  223. raise_error_if_not_found: bool = False,
  224. usecwd: bool = False,
  225. ) -> str:
  226. """
  227. Search in increasingly higher folders for the given file
  228. Returns path to the file if found, or an empty string otherwise
  229. """
  230. def _is_interactive():
  231. """ Decide whether this is running in a REPL or IPython notebook """
  232. main = __import__('__main__', None, None, fromlist=['__file__'])
  233. return not hasattr(main, '__file__')
  234. if usecwd or _is_interactive() or getattr(sys, 'frozen', False):
  235. # Should work without __file__, e.g. in REPL or IPython notebook.
  236. path = os.getcwd()
  237. else:
  238. # will work for .py files
  239. frame = sys._getframe()
  240. current_file = __file__
  241. while frame.f_code.co_filename == current_file:
  242. assert frame.f_back is not None
  243. frame = frame.f_back
  244. frame_filename = frame.f_code.co_filename
  245. path = os.path.dirname(os.path.abspath(frame_filename))
  246. for dirname in _walk_to_root(path):
  247. check_path = os.path.join(dirname, filename)
  248. if os.path.isfile(check_path):
  249. return check_path
  250. if raise_error_if_not_found:
  251. raise IOError('File not found')
  252. return ''
  253. def load_dotenv(
  254. dotenv_path: Optional[StrPath] = None,
  255. stream: Optional[IO[str]] = None,
  256. verbose: bool = False,
  257. override: bool = False,
  258. interpolate: bool = True,
  259. encoding: Optional[str] = "utf-8",
  260. ) -> bool:
  261. """Parse a .env file and then load all the variables found as environment variables.
  262. Parameters:
  263. dotenv_path: Absolute or relative path to .env file.
  264. stream: Text stream (such as `io.StringIO`) with .env content, used if
  265. `dotenv_path` is `None`.
  266. verbose: Whether to output a warning the .env file is missing.
  267. override: Whether to override the system environment variables with the variables
  268. from the `.env` file.
  269. encoding: Encoding to be used to read the file.
  270. Returns:
  271. Bool: True if at least one environment variable is set else False
  272. If both `dotenv_path` and `stream` are `None`, `find_dotenv()` is used to find the
  273. .env file.
  274. """
  275. if dotenv_path is None and stream is None:
  276. dotenv_path = find_dotenv()
  277. dotenv = DotEnv(
  278. dotenv_path=dotenv_path,
  279. stream=stream,
  280. verbose=verbose,
  281. interpolate=interpolate,
  282. override=override,
  283. encoding=encoding,
  284. )
  285. return dotenv.set_as_environment_variables()
  286. def dotenv_values(
  287. dotenv_path: Optional[StrPath] = None,
  288. stream: Optional[IO[str]] = None,
  289. verbose: bool = False,
  290. interpolate: bool = True,
  291. encoding: Optional[str] = "utf-8",
  292. ) -> Dict[str, Optional[str]]:
  293. """
  294. Parse a .env file and return its content as a dict.
  295. The returned dict will have `None` values for keys without values in the .env file.
  296. For example, `foo=bar` results in `{"foo": "bar"}` whereas `foo` alone results in
  297. `{"foo": None}`
  298. Parameters:
  299. dotenv_path: Absolute or relative path to the .env file.
  300. stream: `StringIO` object with .env content, used if `dotenv_path` is `None`.
  301. verbose: Whether to output a warning if the .env file is missing.
  302. encoding: Encoding to be used to read the file.
  303. If both `dotenv_path` and `stream` are `None`, `find_dotenv()` is used to find the
  304. .env file.
  305. """
  306. if dotenv_path is None and stream is None:
  307. dotenv_path = find_dotenv()
  308. return DotEnv(
  309. dotenv_path=dotenv_path,
  310. stream=stream,
  311. verbose=verbose,
  312. interpolate=interpolate,
  313. override=True,
  314. encoding=encoding,
  315. ).dict()