add type annotations

CHANGES.rst

@@ -9,6 +9,7 @@ Unreleased
 -   Bump MarkupSafe dependency to >=1.1.
 -   Bump Babel optional dependency to >=2.1.
 -   Remove code that was marked deprecated.
+-   Add type hinting. :pr:`1412`
 -   Use :pep:`451` API to load templates with
     :class:`~loaders.PackageLoader`. :issue:`1168`
 -   Fix a bug that caused imported macros to not have access to the

MANIFEST.in

@@ -6,4 +6,5 @@ graft docs
 prune docs/_build
 graft examples
 graft tests
+include src/click/py.typed
 global-exclude *.pyc

docs/examples/inline_gettext_extension.py

@@ -30,7 +30,7 @@ def filter_stream(self, stream):
             pos = 0
             lineno = token.lineno
 
-            while 1:
+            while True:
                 if not paren_stack:
                     match = _outside_re.search(token.value, pos)
                 else:

setup.cfg

@@ -86,15 +86,21 @@ per-file-ignores =
 files = src/jinja2
 python_version = 3.6
 disallow_subclassing_any = True
-# disallow_untyped_calls = True
-# disallow_untyped_defs = True
+disallow_untyped_calls = True
+disallow_untyped_defs = True
 disallow_incomplete_defs = True
 no_implicit_optional = True
 local_partial_types = True
-# no_implicit_reexport = True
+no_implicit_reexport = True
 strict_equality = True
 warn_redundant_casts = True
 warn_unused_configs = True
 warn_unused_ignores = True
 warn_return_any = True
 warn_unreachable = True
+
+[mypy-jinja2.defaults]
+no_implicit_reexport = False
+
+[mypy-markupsafe]
+no_implicit_reexport = False

src/jinja2/async_utils.py

@@ -5,27 +5,26 @@
 from .utils import _PassArg
 from .utils import pass_eval_context
 
-if t.TYPE_CHECKING:
-    V = t.TypeVar("V")
+V = t.TypeVar("V")
 
 
-def async_variant(normal_func):
-    def decorator(async_func):
+def async_variant(normal_func):  # type: ignore
+    def decorator(async_func):  # type: ignore
         pass_arg = _PassArg.from_obj(normal_func)
         need_eval_context = pass_arg is None
 
         if pass_arg is _PassArg.environment:
 
-            def is_async(args):
-                return args[0].is_async
+            def is_async(args: t.Any) -> bool:
+                return t.cast(bool, args[0].is_async)
 
         else:
 
-            def is_async(args):
-                return args[0].environment.is_async
+            def is_async(args: t.Any) -> bool:
+                return t.cast(bool, args[0].environment.is_async)
 
         @wraps(normal_func)
-        def wrapper(*args, **kwargs):
+        def wrapper(*args, **kwargs):  # type: ignore
             b = is_async(args)
 
             if need_eval_context:
@@ -45,32 +44,25 @@ def wrapper(*args, **kwargs):
     return decorator
 
 
-async def auto_await(value):
+async def auto_await(value: t.Union[t.Awaitable["V"], "V"]) -> "V":
     if inspect.isawaitable(value):
-        return await value
+        return await t.cast("t.Awaitable[V]", value)
 
-    return value
+    return t.cast("V", value)
 
 
-async def auto_aiter(iterable):
+async def auto_aiter(
+    iterable: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
+) -> "t.AsyncIterator[V]":
     if hasattr(iterable, "__aiter__"):
-        async for item in iterable:
+        async for item in t.cast("t.AsyncIterable[V]", iterable):
             yield item
     else:
-        for item in iterable:
+        for item in t.cast("t.Iterable[V]", iterable):
             yield item
 
 
 async def auto_to_list(
     value: "t.Union[t.AsyncIterable[V], t.Iterable[V]]",
-) -> "t.List[V]":
-    seq = []
-
-    if hasattr(value, "__aiter__"):
-        async for item in t.cast(t.AsyncIterable, value):
-            seq.append(item)
-    else:
-        for item in t.cast(t.Iterable, value):
-            seq.append(item)
-
-    return seq
+) -> t.List["V"]:
+    return [x async for x in auto_aiter(value)]

src/jinja2/bccache.py

@@ -13,10 +13,22 @@
 import stat
 import sys
 import tempfile
+import typing as t
 from hashlib import sha1
 from io import BytesIO
+from types import CodeType
+
+if t.TYPE_CHECKING:
+    import typing_extensions as te
+    from .environment import Environment
+
+    class _MemcachedClient(te.Protocol):
+        def get(self, key: str) -> bytes:
+            ...
+
+        def set(self, key: str, value: bytes, timeout: t.Optional[int] = None) -> None:
+            ...
 
-from .utils import open_if_exists
 
 bc_version = 5
 # Magic bytes to identify Jinja bytecode cache files. Contains the
@@ -38,17 +50,17 @@ class Bucket:
     cache subclasses don't have to care about cache invalidation.
     """
 
-    def __init__(self, environment, key, checksum):
+    def __init__(self, environment: "Environment", key: str, checksum: str) -> None:
         self.environment = environment
         self.key = key
         self.checksum = checksum
         self.reset()
 
-    def reset(self):
+    def reset(self) -> None:
         """Resets the bucket (unloads the bytecode)."""
-        self.code = None
+        self.code: t.Optional[CodeType] = None
 
-    def load_bytecode(self, f):
+    def load_bytecode(self, f: t.BinaryIO) -> None:
         """Loads bytecode from a file or file like object."""
         # make sure the magic header is correct
         magic = f.read(len(bc_magic))
@@ -67,20 +79,20 @@ def load_bytecode(self, f):
             self.reset()
             return
 
-    def write_bytecode(self, f):
+    def write_bytecode(self, f: t.BinaryIO) -> None:
         """Dump the bytecode into the file or file like object passed."""
         if self.code is None:
             raise TypeError("can't write empty bucket")
         f.write(bc_magic)
         pickle.dump(self.checksum, f, 2)
         marshal.dump(self.code, f)
 
-    def bytecode_from_string(self, string):
-        """Load bytecode from a string."""
+    def bytecode_from_string(self, string: bytes) -> None:
+        """Load bytecode from bytes."""
         self.load_bytecode(BytesIO(string))
 
-    def bytecode_to_string(self):
-        """Return the bytecode as string."""
+    def bytecode_to_string(self) -> bytes:
+        """Return the bytecode as bytes."""
         out = BytesIO()
         self.write_bytecode(out)
         return out.getvalue()
@@ -115,41 +127,48 @@ def dump_bytecode(self, bucket):
     Jinja.
     """
 
-    def load_bytecode(self, bucket):
+    def load_bytecode(self, bucket: Bucket) -> None:
         """Subclasses have to override this method to load bytecode into a
         bucket.  If they are not able to find code in the cache for the
         bucket, it must not do anything.
         """
         raise NotImplementedError()
 
-    def dump_bytecode(self, bucket):
+    def dump_bytecode(self, bucket: Bucket) -> None:
         """Subclasses have to override this method to write the bytecode
         from a bucket back to the cache.  If it unable to do so it must not
         fail silently but raise an exception.
         """
         raise NotImplementedError()
 
-    def clear(self):
+    def clear(self) -> None:
         """Clears the cache.  This method is not used by Jinja but should be
         implemented to allow applications to clear the bytecode cache used
         by a particular environment.
         """
 
-    def get_cache_key(self, name, filename=None):
+    def get_cache_key(
+        self, name: str, filename: t.Optional[t.Union[str]] = None
+    ) -> str:
         """Returns the unique hash key for this template name."""
         hash = sha1(name.encode("utf-8"))
+
         if filename is not None:
-            filename = "|" + filename
-            if isinstance(filename, str):
-                filename = filename.encode("utf-8")
-            hash.update(filename)
+            hash.update(f"|{filename}".encode("utf-8"))
+
         return hash.hexdigest()
 
-    def get_source_checksum(self, source):
+    def get_source_checksum(self, source: str) -> str:
         """Returns a checksum for the source."""
         return sha1(source.encode("utf-8")).hexdigest()
 
-    def get_bucket(self, environment, name, filename, source):
+    def get_bucket(
+        self,
+        environment: "Environment",
+        name: str,
+        filename: t.Optional[str],
+        source: str,
+    ) -> Bucket:
         """Return a cache bucket for the given template.  All arguments are
         mandatory but filename may be `None`.
         """
@@ -159,7 +178,7 @@ def get_bucket(self, environment, name, filename, source):
         self.load_bytecode(bucket)
         return bucket
 
-    def set_bucket(self, bucket):
+    def set_bucket(self, bucket: Bucket) -> None:
         """Put the bucket into the cache."""
         self.dump_bytecode(bucket)
 
@@ -182,14 +201,16 @@ class FileSystemBytecodeCache(BytecodeCache):
     This bytecode cache supports clearing of the cache using the clear method.
     """
 
-    def __init__(self, directory=None, pattern="__jinja2_%s.cache"):
+    def __init__(
+        self, directory: t.Optional[str] = None, pattern: str = "__jinja2_%s.cache"
+    ) -> None:
         if directory is None:
             directory = self._get_default_cache_dir()
         self.directory = directory
         self.pattern = pattern
 
-    def _get_default_cache_dir(self):
-        def _unsafe_dir():
+    def _get_default_cache_dir(self) -> str:
+        def _unsafe_dir() -> t.NoReturn:
             raise RuntimeError(
                 "Cannot determine safe temp directory.  You "
                 "need to explicitly provide one."
@@ -235,25 +256,21 @@ def _unsafe_dir():
 
         return actual_dir
 
-    def _get_cache_filename(self, bucket):
+    def _get_cache_filename(self, bucket: Bucket) -> str:
         return os.path.join(self.directory, self.pattern % (bucket.key,))
 
-    def load_bytecode(self, bucket):
-        f = open_if_exists(self._get_cache_filename(bucket), "rb")
-        if f is not None:
-            try:
+    def load_bytecode(self, bucket: Bucket) -> None:
+        filename = self._get_cache_filename(bucket)
+
+        if os.path.exists(filename):
+            with open(filename, "rb") as f:
                 bucket.load_bytecode(f)
-            finally:
-                f.close()
 
-    def dump_bytecode(self, bucket):
-        f = open(self._get_cache_filename(bucket), "wb")
-        try:
+    def dump_bytecode(self, bucket: Bucket) -> None:
+        with open(self._get_cache_filename(bucket), "wb") as f:
             bucket.write_bytecode(f)
-        finally:
-            f.close()
 
-    def clear(self):
+    def clear(self) -> None:
         # imported lazily here because google app-engine doesn't support
         # write access on the file system and the function does not exist
         # normally.
@@ -314,32 +331,34 @@ class MemcachedBytecodeCache(BytecodeCache):
 
     def __init__(
         self,
-        client,
-        prefix="jinja2/bytecode/",
-        timeout=None,
-        ignore_memcache_errors=True,
+        client: "_MemcachedClient",
+        prefix: str = "jinja2/bytecode/",
+        timeout: t.Optional[int] = None,
+        ignore_memcache_errors: bool = True,
     ):
         self.client = client
         self.prefix = prefix
         self.timeout = timeout
         self.ignore_memcache_errors = ignore_memcache_errors
 
-    def load_bytecode(self, bucket):
+    def load_bytecode(self, bucket: Bucket) -> None:
         try:
             code = self.client.get(self.prefix + bucket.key)
         except Exception:
             if not self.ignore_memcache_errors:
                 raise
-            code = None
-        if code is not None:
+        else:
             bucket.bytecode_from_string(code)
 
-    def dump_bytecode(self, bucket):
-        args = (self.prefix + bucket.key, bucket.bytecode_to_string())
-        if self.timeout is not None:
-            args += (self.timeout,)
+    def dump_bytecode(self, bucket: Bucket) -> None:
+        key = self.prefix + bucket.key
+        value = bucket.bytecode_to_string()
+
         try:
-            self.client.set(*args)
+            if self.timeout is not None:
+                self.client.set(key, value, self.timeout)
+            else:
+                self.client.set(key, value)
         except Exception:
             if not self.ignore_memcache_errors:
                 raise