99import sys
1010from subprocess import Popen
1111import tempfile
12+ from typing import (Any , Dict , Iterator , List , Match , NamedTuple , Optional , # noqa
13+ Pattern , Union , TYPE_CHECKING , Text , IO , Tuple ) # noqa
1214import warnings
13- from collections import OrderedDict , namedtuple
15+ from collections import OrderedDict
1416from contextlib import contextmanager
1517
1618from .compat import StringIO , PY2 , WIN , text_type
1719
18- __posix_variable = re .compile (r'\$\{[^\}]*\}' )
20+ if TYPE_CHECKING : # pragma: no cover
21+ if sys .version_info >= (3 , 6 ):
22+ _PathLike = os .PathLike
23+ else :
24+ _PathLike = Text
25+
26+ if sys .version_info >= (3 , 0 ):
27+ _StringIO = StringIO
28+ else :
29+ _StringIO = StringIO [Text ]
30+
31+ __posix_variable = re .compile (r'\$\{[^\}]*\}' ) # type: Pattern[Text]
1932
2033_binding = re .compile (
2134 r"""
4255 )
4356 """ .format (r'[^\S\r\n]' ),
4457 re .MULTILINE | re .VERBOSE ,
45- )
58+ ) # type: Pattern[Text]
4659
47- _escape_sequence = re .compile (r"\\[\\'\"abfnrtv]" )
60+ _escape_sequence = re .compile (r"\\[\\'\"abfnrtv]" ) # type: Pattern[Text]
4861
4962
50- Binding = namedtuple ('Binding' , 'key value original' )
63+ Binding = NamedTuple ("Binding" , [("key" , Optional [Text ]),
64+ ("value" , Optional [Text ]),
65+ ("original" , Text )])
5166
5267
5368def decode_escapes (string ):
69+ # type: (Text) -> Text
5470 def decode_match (match ):
55- return codecs .decode (match .group (0 ), 'unicode-escape' )
71+ # type: (Match[Text]) -> Text
72+ return codecs .decode (match .group (0 ), 'unicode-escape' ) # type: ignore
5673
5774 return _escape_sequence .sub (decode_match , string )
5875
5976
6077def is_surrounded_by (string , char ):
78+ # type: (Text, Text) -> bool
6179 return (
6280 len (string ) > 1
6381 and string [0 ] == string [- 1 ] == char
6482 )
6583
6684
6785def parse_binding (string , position ):
86+ # type: (Text, int) -> Tuple[Binding, int]
6887 match = _binding .match (string , position )
88+ assert match is not None
6989 (matched , key , value ) = match .groups ()
7090 if key is None or value is None :
7191 key = None
@@ -80,6 +100,7 @@ def parse_binding(string, position):
80100
81101
82102def parse_stream (stream ):
103+ # type:(IO[Text]) -> Iterator[Binding]
83104 string = stream .read ()
84105 position = 0
85106 length = len (string )
@@ -91,23 +112,26 @@ def parse_stream(stream):
91112class DotEnv ():
92113
93114 def __init__ (self , dotenv_path , verbose = False ):
94- self .dotenv_path = dotenv_path
95- self ._dict = None
96- self .verbose = verbose
115+ # type: (Union[Text, _PathLike, _StringIO], bool) -> None
116+ self .dotenv_path = dotenv_path # type: Union[Text,_PathLike, _StringIO]
117+ self ._dict = None # type: Optional[Dict[Text, Text]]
118+ self .verbose = verbose # type: bool
97119
98120 @contextmanager
99121 def _get_stream (self ):
122+ # type: () -> Iterator[IO[Text]]
100123 if isinstance (self .dotenv_path , StringIO ):
101124 yield self .dotenv_path
102125 elif os .path .isfile (self .dotenv_path ):
103126 with io .open (self .dotenv_path ) as stream :
104127 yield stream
105128 else :
106129 if self .verbose :
107- warnings .warn ("File doesn't exist {}" .format (self .dotenv_path ))
130+ warnings .warn ("File doesn't exist {}" .format (self .dotenv_path )) # type: ignore
108131 yield StringIO ('' )
109132
110133 def dict (self ):
134+ # type: () -> Dict[Text, Text]
111135 """Return dotenv as dict"""
112136 if self ._dict :
113137 return self ._dict
@@ -117,12 +141,14 @@ def dict(self):
117141 return self ._dict
118142
119143 def parse (self ):
144+ # type: () -> Iterator[Tuple[Text, Text]]
120145 with self ._get_stream () as stream :
121146 for mapping in parse_stream (stream ):
122147 if mapping .key is not None and mapping .value is not None :
123148 yield mapping .key , mapping .value
124149
125150 def set_as_environment_variables (self , override = False ):
151+ # type: (bool) -> bool
126152 """
127153 Load the current dotenv as system environemt variable.
128154 """
@@ -135,11 +161,12 @@ def set_as_environment_variables(self, override=False):
135161 if isinstance (k , text_type ) or isinstance (v , text_type ):
136162 k = k .encode ('ascii' )
137163 v = v .encode ('ascii' )
138- os .environ [k ] = v
164+ os .environ [k ] = v # type: ignore
139165
140166 return True
141167
142168 def get (self , key ):
169+ # type: (Text) -> Optional[Text]
143170 """
144171 """
145172 data = self .dict ()
@@ -148,10 +175,13 @@ def get(self, key):
148175 return data [key ]
149176
150177 if self .verbose :
151- warnings .warn ("key %s not found in %s." % (key , self .dotenv_path ))
178+ warnings .warn ("key %s not found in %s." % (key , self .dotenv_path )) # type: ignore
179+
180+ return None
152181
153182
154183def get_key (dotenv_path , key_to_get ):
184+ # type: (Union[Text, _PathLike], Text) -> Optional[Text]
155185 """
156186 Gets the value of a given key from the given .env
157187
@@ -162,10 +192,11 @@ def get_key(dotenv_path, key_to_get):
162192
163193@contextmanager
164194def rewrite (path ):
195+ # type: (_PathLike) -> Iterator[Tuple[IO[Text], IO[Text]]]
165196 try :
166197 with tempfile .NamedTemporaryFile (mode = "w+" , delete = False ) as dest :
167198 with io .open (path ) as source :
168- yield (source , dest )
199+ yield (source , dest ) # type: ignore
169200 except BaseException :
170201 if os .path .isfile (dest .name ):
171202 os .unlink (dest .name )
@@ -175,6 +206,7 @@ def rewrite(path):
175206
176207
177208def set_key (dotenv_path , key_to_set , value_to_set , quote_mode = "always" ):
209+ # type: (_PathLike, Text, Text, Text) -> Tuple[Optional[bool], Text, Text]
178210 """
179211 Adds or Updates a key/value to the given .env
180212
@@ -183,7 +215,7 @@ def set_key(dotenv_path, key_to_set, value_to_set, quote_mode="always"):
183215 """
184216 value_to_set = value_to_set .strip ("'" ).strip ('"' )
185217 if not os .path .exists (dotenv_path ):
186- warnings .warn ("can't write to %s - it doesn't exist." % dotenv_path )
218+ warnings .warn ("can't write to %s - it doesn't exist." % dotenv_path ) # type: ignore
187219 return None , key_to_set , value_to_set
188220
189221 if " " in value_to_set :
@@ -207,14 +239,15 @@ def set_key(dotenv_path, key_to_set, value_to_set, quote_mode="always"):
207239
208240
209241def unset_key (dotenv_path , key_to_unset , quote_mode = "always" ):
242+ # type: (_PathLike, Text, Text) -> Tuple[Optional[bool], Text]
210243 """
211244 Removes a given key from the given .env
212245
213246 If the .env path given doesn't exist, fails
214247 If the given key doesn't exist in the .env, fails
215248 """
216249 if not os .path .exists (dotenv_path ):
217- warnings .warn ("can't delete from %s - it doesn't exist." % dotenv_path )
250+ warnings .warn ("can't delete from %s - it doesn't exist." % dotenv_path ) # type: ignore
218251 return None , key_to_unset
219252
220253 removed = False
@@ -226,14 +259,16 @@ def unset_key(dotenv_path, key_to_unset, quote_mode="always"):
226259 dest .write (mapping .original )
227260
228261 if not removed :
229- warnings .warn ("key %s not removed from %s - key doesn't exist." % (key_to_unset , dotenv_path ))
262+ warnings .warn ("key %s not removed from %s - key doesn't exist." % (key_to_unset , dotenv_path )) # type: ignore
230263 return None , key_to_unset
231264
232265 return removed , key_to_unset
233266
234267
235268def resolve_nested_variables (values ):
269+ # type: (Dict[Text, Text]) -> Dict[Text, Text]
236270 def _replacement (name ):
271+ # type: (Text) -> Text
237272 """
238273 get appropriate value for a variable name.
239274 first search in environ, if not found,
@@ -243,6 +278,7 @@ def _replacement(name):
243278 return ret
244279
245280 def _re_sub_callback (match_object ):
281+ # type: (Match[Text]) -> Text
246282 """
247283 From a match object gets the variable name and returns
248284 the correct replacement
@@ -258,6 +294,7 @@ def _re_sub_callback(match_object):
258294
259295
260296def _walk_to_root (path ):
297+ # type: (Text) -> Iterator[Text]
261298 """
262299 Yield directories starting from the given directory up to the root
263300 """
@@ -276,6 +313,7 @@ def _walk_to_root(path):
276313
277314
278315def find_dotenv (filename = '.env' , raise_error_if_not_found = False , usecwd = False ):
316+ # type: (Text, bool, bool) -> Text
279317 """
280318 Search in increasingly higher folders for the given file
281319
@@ -312,16 +350,19 @@ def find_dotenv(filename='.env', raise_error_if_not_found=False, usecwd=False):
312350
313351
314352def load_dotenv (dotenv_path = None , stream = None , verbose = False , override = False ):
353+ # type: (Union[Text, _PathLike, None], Optional[_StringIO], bool, bool) -> bool
315354 f = dotenv_path or stream or find_dotenv ()
316355 return DotEnv (f , verbose = verbose ).set_as_environment_variables (override = override )
317356
318357
319358def dotenv_values (dotenv_path = None , stream = None , verbose = False ):
359+ # type: (Union[Text, _PathLike, None], Optional[_StringIO], bool) -> Dict[Text, Text]
320360 f = dotenv_path or stream or find_dotenv ()
321361 return DotEnv (f , verbose = verbose ).dict ()
322362
323363
324364def run_command (command , env ):
365+ # type: (List[str], Dict[str, str]) -> int
325366 """Run command in sub process.
326367
327368 Runs the command in a sub process with the variables from `env`
0 commit comments