Repository: quora/pyanalyze
Pull request #673 (2023-09-10 00:43:17)

Support for annotated_types
This adds support for checking `annotated_types` annotations against literals, as well as a few tricks to make them more usable:
- Allowing stronger annotations where a weaker one is requested; for example, allow `Annotated[int, Gt(6)]` where `Annotated[int, Gt(5)]` is requested.
- Inferring annotations in some cases (`if x > 5:` essentially adds the annotation `Gt(5)`).

Future work:
- We should also infer predicates for `if len(x) > 5:` but we currently don't.


Patch:
diff --git a/docs/annotated_types.md b/docs/annotated_types.md
new file mode 100644
index 0000000..a789391
--- /dev/null
+++ b/docs/annotated_types.md
@@ -0,0 +1,44 @@
+# Support for `annotated_types`
+
+Pyanalyze supports the [annotated-types](https://pypi.org/project/annotated-types/) library, which provides a set of common primitives to use in `Annotated` metadata.
+
+This is useful for restricting the value of an object:
+
+```python
+from typing_extensions import Annotated
+from annotated_types import Gt
+
+def takes_gt_5(x: Annotated[int, Gt(5)]) -> None:
+    assert x > 5, "number too small"
+
+def caller() -> None:
+    takes_gt_5(6)  # ok
+    takes_gt_5(5)  # type checker error
+```
+
+Pyanalyze enforces these annotations strictly: if it cannot determine whether or
+not a value fulfills the predicate, it shows an error. For example, the following
+will be rejected:
+
+```python
+def caller(i: int) -> None:
+    takes_gt_5(i)  # type checker error, as it may be less than 5
+```
+
+## Notes on specific predicates
+
+Pyanalyze infers the interval attributes `Gt`, `Ge`, `Lt`, and `Le` based
+on comparisons with literals:
+
+```python
+def caller(i: int) -> None:
+    takes_gt_5(i)  # error
+
+    if i > 5:
+        takes_gt_5(i)  # accepted
+```
+
+For the `MultipleOf` check, pyanalyze follows Python semantics: values
+are accepted if `value % multiple_of == 0`.
+
+For the `Timezone` check, support for requiring string-based timezones is not implemented.
diff --git a/docs/changelog.md b/docs/changelog.md
index 59807f9..50ba9d6 100644
--- a/docs/changelog.md
+++ b/docs/changelog.md
@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Support for annotations from the `annotated-types` library (#673)
 - Detect undefined attributes on Pydantic models (#670)
 - Remove duplicate "attribute_is_never_set" error for classes
   with predefined attributes (#670)
diff --git a/docs/index.rst b/docs/index.rst
index 029743a..4af9059 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -15,6 +15,7 @@ pyanalyze: A semi-static typechecker
    typesystem
    design
    type_evaluation
+   annotated_types
    glossary
    reference/annotations
    reference/ast_annotator
diff --git a/pyanalyze/__init__.py b/pyanalyze/__init__.py
index de700d6..a750ccc 100644
--- a/pyanalyze/__init__.py
+++ b/pyanalyze/__init__.py
@@ -8,6 +8,7 @@ pyanalyze is a package for Python static analysis.
 
 from . import (
     analysis_lib,
+    annotated_types,
     annotations,
     arg_spec,
     ast_annotator,
@@ -64,3 +65,4 @@ used(shared_options)
 used(functions)
 used(predicates)
 used(typevar)
+used(annotated_types)
diff --git a/pyanalyze/annotated_types.py b/pyanalyze/annotated_types.py
new file mode 100644
index 0000000..dc4411d
--- /dev/null
+++ b/pyanalyze/annotated_types.py
@@ -0,0 +1,303 @@
+"""
+
+Support for annotations from the annotated_types library.
+
+"""
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any, Callable, Iterable, Optional, Union
+
+from pyanalyze.value import CanAssign, CanAssignContext, Value, flatten_values
+
+from .extensions import CustomCheck
+from .value import (
+    AnnotatedValue,
+    AnyValue,
+    CanAssignError,
+    CustomCheckExtension,
+    DictIncompleteValue,
+    KnownValue,
+    SequenceValue,
+    TypedDictValue,
+    unannotate,
+)
+
+try:
+    import annotated_types
+except ImportError:
+
+    def get_annotated_types_extension(obj: object) -> Iterable[CustomCheckExtension]:
+        return []
+
+else:
+
+    def get_annotated_types_extension(obj: object) -> Iterable[CustomCheckExtension]:
+        if isinstance(obj, annotated_types.GroupedMetadata):
+            for value in obj:
+                maybe_ext = _get_single_annotated_types_extension(value)
+                if maybe_ext is not None:
+                    yield CustomCheckExtension(maybe_ext)
+        elif isinstance(obj, annotated_types.BaseMetadata):
+            maybe_ext = _get_single_annotated_types_extension(obj)
+            if maybe_ext is not None:
+                yield CustomCheckExtension(maybe_ext)
+
+    def _get_single_annotated_types_extension(
+        obj: annotated_types.BaseMetadata,
+    ) -> Optional[CustomCheck]:
+        if isinstance(obj, annotated_types.Gt):
+            return Gt(obj.gt)
+        elif isinstance(obj, annotated_types.Ge):
+            return Ge(obj.ge)
+        elif isinstance(obj, annotated_types.Lt):
+            return Lt(obj.lt)
+        elif isinstance(obj, annotated_types.Le):
+            return Le(obj.le)
+        elif isinstance(obj, annotated_types.MultipleOf):
+            return MultipleOf(obj.multiple_of)
+        elif isinstance(obj, annotated_types.MinLen):
+            return MinLen(obj.min_length)
+        elif isinstance(obj, annotated_types.MaxLen):
+            return MaxLen(obj.max_length)
+        elif isinstance(obj, annotated_types.Timezone):
+            return Timezone(obj.tz)
+        elif isinstance(obj, annotated_types.Predicate):
+            return Predicate(obj.func)
+        else:
+            # In case annotated_types adds more kinds of checks in the future
+            return None
+
+
+@dataclass(frozen=True)
+class AnnotatedTypesCheck(CustomCheck):
+    def can_assign(self, value: Value, ctx: CanAssignContext) -> CanAssign:
+        for subval in flatten_values(value):
+            original_subval = subval
+            if isinstance(subval, AnnotatedValue):
+                if any(
+                    ext == self or self.is_compatible_metadata(ext)
+                    for ext in subval.get_custom_check_of_type(AnnotatedTypesCheck)
+                ):
+                    continue
+            subval = unannotate(subval)
+            if isinstance(subval, AnyValue):
+                continue
+            if isinstance(subval, KnownValue):
+                try:
+                    result = self.predicate(subval.val)
+                except Exception:
+                    return CanAssignError(
+                        f"Value {subval.val} cannot be checked against predicate {self}"
+                    )
+                else:
+                    if not result:
+                        return CanAssignError(
+                            f"Value {subval.val} does not match predicate {self}"
+                        )
+            else:
+                can_assign = self.can_assign_non_literal(original_subval)
+                if isinstance(can_assign, CanAssignError):
+                    return can_assign
+        return {}
+
+    def predicate(self, value: Any) -> bool:
+        raise NotImplementedError
+
+    def is_compatible_metadata(self, metadata: "AnnotatedTypesCheck") -> bool:
+        """Override this to allow metadata that is not exactly the same as the
+        one in this object to match. For example, Gt(5) should accept Gt(4), as that
+        is a strictly weaker condition.
+        """
+        return False
+
+    def can_assign_non_literal(self, value: Value) -> CanAssign:
+        """Override this to allow some non-Literal values. For example, the
+        MinLen extensions can allow sequences with sufficiently known lengths."""
+        return CanAssignError(
+            f"Cannot determine whether {value} fulfills predicate {self}"
+        )
+
+
+@dataclass(frozen=True)
+class Gt(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return value > self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, Gt):
+            return metadata.value >= self.value
+        elif isinstance(metadata, Ge):
+            return metadata.value > self.value
+        else:
+            return False
+
+
+@dataclass(frozen=True)
+class Ge(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return value >= self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, Gt):
+            return metadata.value >= self.value
+        elif isinstance(metadata, Ge):
+            return metadata.value >= self.value
+        else:
+            return False
+
+
+@dataclass(frozen=True)
+class Lt(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return value < self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, Lt):
+            return metadata.value <= self.value
+        elif isinstance(metadata, Le):
+            return metadata.value < self.value
+        else:
+            return False
+
+
+@dataclass(frozen=True)
+class Le(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return value <= self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, Lt):
+            return metadata.value <= self.value
+        elif isinstance(metadata, Le):
+            return metadata.value <= self.value
+        else:
+            return False
+
+
+@dataclass(frozen=True)
+class MultipleOf(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return value % self.value == 0
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, MultipleOf):
+            # If we want a MultipleOf(5), but we're passed a MultipleOf(10), that's ok
+            return metadata.value % self.value == 0
+        else:
+            return False
+
+
+@dataclass(frozen=True)
+class MinLen(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return len(value) >= self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, MinLen):
+            return metadata.value >= self.value
+        else:
+            return False
+
+    def can_assign_non_literal(self, value: Value) -> CanAssign:
+        min_len = _min_len_of_value(value)
+        if min_len is not None and min_len >= self.value:
+            return {}
+        return super().can_assign_non_literal(value)
+
+
+@dataclass(frozen=True)
+class MaxLen(AnnotatedTypesCheck):
+    value: Any
+
+    def predicate(self, value: Any) -> bool:
+        return len(value) <= self.value
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if isinstance(metadata, MaxLen):
+            return metadata.value <= self.value
+        else:
+            return False
+
+    def can_assign_non_literal(self, value: Value) -> CanAssign:
+        max_len = _max_len_of_value(value)
+        if max_len is not None and max_len <= self.value:
+            return {}
+        return super().can_assign_non_literal(value)
+
+
+@dataclass(frozen=True)
+class Timezone(AnnotatedTypesCheck):
+    value: Union[str, timezone, type(...), None]
+
+    def predicate(self, value: Any) -> bool:
+        if not isinstance(value, datetime):
+            return False
+        if self.value is None:
+            return value.tzinfo is None
+        elif self.value is ...:
+            return value.tzinfo is not None
+        elif isinstance(self.value, timezone):
+            return value.tzinfo == self.value
+        else:
+            return False
+
+    def is_compatible_metadata(self, metadata: AnnotatedTypesCheck) -> bool:
+        if (
+            self.value is ...
+            and isinstance(metadata, Timezone)
+            and metadata.value is not None
+        ):
+            return True
+        return False
+
+
+@dataclass(frozen=True)
+class Predicate(AnnotatedTypesCheck):
+    predicate_callable: Callable[[Any], bool]
+
+    def predicate(self, value: Any) -> bool:
+        return self.predicate_callable(value)
+
+
+def _min_len_of_value(val: Value) -> Optional[int]:
+    if isinstance(val, SequenceValue):
+        return sum(is_many is False for is_many, _ in val.members)
+    elif isinstance(val, DictIncompleteValue):
+        return sum(pair.is_required and not pair.is_many for pair in val.kv_pairs)
+    elif isinstance(val, TypedDictValue):
+        return sum(required for required, _ in val.items.values())
+    else:
+        return None
+
+
+def _max_len_of_value(val: Value) -> Optional[int]:
+    if isinstance(val, SequenceValue):
+        maximum = 0
+        for is_many, _ in val.members:
+            if is_many:
+                return None
+            maximum += 1
+        return maximum
+    elif isinstance(val, DictIncompleteValue):
+        maximum = 0
+        for pair in val.kv_pairs:
+            if pair.is_many:
+                return None
+            if pair.is_required:
+                maximum += 1
+        return maximum
+    else:
+        # Always None for TypedDicts as TypedDicts may have arbitrary extra keys
+        return None
diff --git a/pyanalyze/annotations.py b/pyanalyze/annotations.py
index c3edd76..912301a 100644
--- a/pyanalyze/annotations.py
+++ b/pyanalyze/annotations.py
@@ -23,7 +23,6 @@ These functions all use :class:`Context` objects to resolve names and
 show errors.
 
 """
-from _ast import Dict
 import ast
 import builtins
 import contextlib
@@ -36,6 +35,7 @@ from typing import (
     Container,
     ContextManager,
     Generator,
+    List,
     Mapping,
     NewType,
     Optional,
@@ -51,6 +51,8 @@ import qcore
 
 from typing_extensions import ParamSpec, TypedDict, get_origin, get_args
 
+from pyanalyze.annotated_types import get_annotated_types_extension
+
 from . import type_evaluation
 
 from .error_code import ErrorCode
@@ -924,7 +926,7 @@ class _Visitor(ast.NodeVisitor):
         elts = [(False, self.visit(elt)) for elt in node.elts]
         return SequenceValue(set, elts)
 
-    def visit_Dict(self, node: Dict) -> Any:
+    def visit_Dict(self, node: ast.Dict) -> Any:
         keys = [self.visit(key) if key is not None else None for key in node.keys]
         values = [self.visit(value) for value in node.values]
         kvpairs = []
@@ -1278,29 +1280,43 @@ def _make_callable_from_value(
 
 
 def _make_annotated(origin: Value, metadata: Sequence[Value], ctx: Context) -> Value:
-    metadata = [_value_from_metadata(entry, ctx) for entry in metadata]
-    return annotate_value(origin, metadata)
-
-
-def _value_from_metadata(entry: Value, ctx: Context) -> Union[Value, Extension]:
-    if isinstance(entry, KnownValue):
-        if isinstance(entry.val, ParameterTypeGuard):
-            return ParameterTypeGuardExtension(
-                entry.val.varname, _type_from_runtime(entry.val.guarded_type, ctx)
-            )
-        elif isinstance(entry.val, NoReturnGuard):
-            return NoReturnGuardExtension(
-                entry.val.varname, _type_from_runtime(entry.val.guarded_type, ctx)
-            )
-        elif isinstance(entry.val, HasAttrGuard):
-            return HasAttrGuardExtension(
-                entry.val.varname,
-                _type_from_runtime(entry.val.attribute_name, ctx),
-                _type_from_runtime(entry.val.attribute_type, ctx),
-            )
-        elif isinstance(entry.val, CustomCheck):
-            return CustomCheckExtension(entry.val)
-    return entry
+    metadata_objs: List[Union[Value, Extension]] = []
+    for entry in metadata:
+        if isinstance(entry, KnownValue):
+            if isinstance(entry.val, ParameterTypeGuard):
+                metadata_objs.append(
+                    ParameterTypeGuardExtension(
+                        entry.val.varname,
+                        _type_from_runtime(entry.val.guarded_type, ctx),
+                    )
+                )
+                continue
+            elif isinstance(entry.val, NoReturnGuard):
+                metadata_objs.append(
+                    NoReturnGuardExtension(
+                        entry.val.varname,
+                        _type_from_runtime(entry.val.guarded_type, ctx),
+                    )
+                )
+                continue
+            elif isinstance(entry.val, HasAttrGuard):
+                metadata_objs.append(
+                    HasAttrGuardExtension(
+                        entry.val.varname,
+                        _type_from_runtime(entry.val.attribute_name, ctx),
+                        _type_from_runtime(entry.val.attribute_type, ctx),
+                    )
+                )
+                continue
+            elif isinstance(entry.val, CustomCheck):
+                metadata_objs.append(CustomCheckExtension(entry.val))
+                continue
+            annotated_types_extensions = list(get_annotated_types_extension(entry.val))
+            if annotated_types_extensions:
+                metadata_objs.extend(annotated_types_extensions)
+            else:
+                metadata_objs.append(entry)
+    return annotate_value(origin, metadata_objs)
 
 
 _CONTEXT_MANAGER_TYPES = {
diff --git a/pyanalyze/name_check_visitor.py b/pyanalyze/name_check_visitor.py
index b31175e..bb45b51 100644
--- a/pyanalyze/name_check_visitor.py
+++ b/pyanalyze/name_check_visitor.py
@@ -52,6 +52,8 @@ import qcore
 import typeshed_client
 from typing_extensions import Annotated, Protocol, get_args, get_origin
 
+from pyanalyze.annotated_types import Gt, Ge, Le, Lt
+
 from . import attributes, format_strings, importer, node_visitor, type_evaluation
 from .analysis_lib import get_attribute_path
 from .annotations import (
@@ -156,6 +158,7 @@ from .value import (
     SYS_PLATFORM_EXTENSION,
     SYS_VERSION_INFO_EXTENSION,
     AlwaysPresentExtension,
+    CustomCheckExtension,
     DefiniteValueExtension,
     DeprecatedExtension,
     SkipDeprecatedExtension,
@@ -293,16 +296,16 @@ def _not_in(a: object, b: Container[object]) -> bool:
 
 
 COMPARATOR_TO_OPERATOR = {
-    ast.Eq: (operator.eq, operator.ne),
-    ast.NotEq: (operator.ne, operator.eq),
-    ast.Lt: (operator.lt, operator.ge),
-    ast.LtE: (operator.le, operator.gt),
-    ast.Gt: (operator.gt, operator.le),
-    ast.GtE: (operator.ge, operator.lt),
-    ast.Is: (operator.is_, operator.is_not),
-    ast.IsNot: (operator.is_not, operator.is_),
-    ast.In: (_in, _not_in),
-    ast.NotIn: (_not_in, _in),
+    ast.Eq: (operator.eq, operator.ne, None),
+    ast.NotEq: (operator.ne, operator.eq, None),
+    ast.Lt: (operator.lt, operator.ge, Lt),
+    ast.LtE: (operator.le, operator.gt, Le),
+    ast.Gt: (operator.gt, operator.le, Gt),
+    ast.GtE: (operator.ge, operator.lt, Ge),
+    ast.Is: (operator.is_, operator.is_not, None),
+    ast.IsNot: (operator.is_not, operator.is_, None),
+    ast.In: (_in, _not_in, None),
+    ast.NotIn: (_not_in, _in, None),
 }
 
 SAFE_DECORATORS_FOR_ARGSPEC_TO_RETVAL = [KnownValue(asynq.asynq), KnownValue(property)]
@@ -3242,14 +3245,14 @@ class NameCheckVisitor(node_visitor.ReplacingNodeVisitor):
                 and isinstance(rhs, KnownValue)
                 and isinstance(op, (ast.Eq, ast.NotEq))
             ):
-                op_func, _ = COMPARATOR_TO_OPERATOR[type(op)]
+                op_func, _, _ = COMPARATOR_TO_OPERATOR[type(op)]
                 definite_value = op_func(sys.platform, rhs.val)
             elif (
                 SYS_VERSION_INFO_EXTENSION in lhs.metadata
                 and isinstance(rhs, KnownValue)
                 and isinstance(op, (ast.Gt, ast.GtE, ast.Lt, ast.LtE))
             ):
-                op_func, _ = COMPARATOR_TO_OPERATOR[type(op)]
+                op_func, _, _ = COMPARATOR_TO_OPERATOR[type(op)]
                 definite_value = op_func(sys.version_info, rhs.val)
             lhs = lhs.value
         if isinstance(lhs_constraint, PredicateProvider) and isinstance(
@@ -3332,7 +3335,7 @@ class NameCheckVisitor(node_visitor.ReplacingNodeVisitor):
             positive = isinstance(op, ast.In)
             return Constraint(varname, ConstraintType.predicate, positive, predicate)
         else:
-            positive_operator, negative_operator = COMPARATOR_TO_OPERATOR[type(op)]
+            positive_operator, negative_operator, ext = COMPARATOR_TO_OPERATOR[type(op)]
 
             def predicate_func(value: Value, positive: bool) -> Optional[Value]:
                 op = positive_operator if positive else negative_operator
@@ -3347,6 +3350,9 @@ class NameCheckVisitor(node_visitor.ReplacingNodeVisitor):
                     else:
                         if not result:
                             return None
+                    return value
+                if ext is not None and positive:
+                    return annotate_value(value, [CustomCheckExtension(ext(other_val))])
                 return value
 
             return Constraint(varname, ConstraintType.predicate, True, predicate_func)
@@ -3354,7 +3360,7 @@ class NameCheckVisitor(node_visitor.ReplacingNodeVisitor):
     def _constraint_from_predicate_provider(
         self, pred: PredicateProvider, other_val: Any, op: ast.AST
     ) -> Constraint:
-        positive_operator, negative_operator = COMPARATOR_TO_OPERATOR[type(op)]
+        positive_operator, negative_operator, _ = COMPARATOR_TO_OPERATOR[type(op)]
 
         def predicate_func(value: Value, positive: bool) -> Optional[Value]:
             predicate_value = pred.provider(value)
diff --git a/setup.py b/setup.py
index 38c9c03..6a47215 100644
--- a/setup.py
+++ b/setup.py
@@ -39,7 +39,15 @@ if __name__ == "__main__":
             "codemod",
             "tomli>=1.1.0",
         ],
-        extras_require={"tests": ["pytest", "mypy_extensions", "attrs", "pydantic"]},
+        extras_require={
+            "tests": [
+                "pytest",
+                "mypy_extensions",
+                "attrs",
+                "pydantic",
+                "annotated-types",
+            ]
+        },
         # These are useful for unit tests of pyanalyze extensions
         # outside the package.
         package_data={"pyanalyze": package_data},


Test patch:
diff --git a/pyanalyze/test_annotated_types.py b/pyanalyze/test_annotated_types.py
new file mode 100644
index 0000000..cb14b31
--- /dev/null
+++ b/pyanalyze/test_annotated_types.py
@@ -0,0 +1,281 @@
+# static analysis: ignore
+from .test_name_check_visitor import TestNameCheckVisitorBase
+from .test_node_visitor import assert_passes
+
+
+class TestAnnotatedTypesAnnotations(TestNameCheckVisitorBase):
+    @assert_passes()
+    def test_gt(self):
+        from typing_extensions import Annotated
+        from typing import Any
+        from annotated_types import Gt, Ge
+
+        def takes_gt_5(x: Annotated[Any, Gt(5)]) -> None:
+            pass
+
+        def capybara(
+            i: int,
+            unnannotated,
+            gt_4: Annotated[int, Gt(4)],
+            gt_5: Annotated[int, Gt(5)],
+            gt_6: Annotated[int, Gt(6)],
+            ge_5: Annotated[int, Ge(5)],
+            ge_6: Annotated[int, Ge(6)],
+        ) -> None:
+            takes_gt_5(i)  # E: incompatible_argument
+            takes_gt_5(unnannotated)
+            takes_gt_5(gt_4)  # E: incompatible_argument
+            takes_gt_5(gt_5)
+            takes_gt_5(gt_6)
+            takes_gt_5(ge_5)  # E: incompatible_argument
+            takes_gt_5(ge_6)
+            takes_gt_5(5)  # E: incompatible_argument
+            takes_gt_5(6)
+            takes_gt_5("not an int")  # E: incompatible_argument
+
+    @assert_passes()
+    def test_ge(self):
+        from typing_extensions import Annotated
+        from typing import Any
+        from annotated_types import Ge, Gt
+
+        def takes_ge_5(x: Annotated[Any, Ge(5)]) -> None:
+            pass
+
+        def capybara(
+            i: int,
+            unnannotated,
+            ge_4: Annotated[int, Ge(4)],
+            ge_5: Annotated[int, Ge(5)],
+            ge_6: Annotated[int, Ge(6)],
+            gt_4: Annotated[int, Gt(4)],
+            gt_5: Annotated[int, Gt(5)],
+        ) -> None:
+            takes_ge_5(i)  # E: incompatible_argument
+            takes_ge_5(unnannotated)
+            takes_ge_5(ge_4)  # E: incompatible_argument
+            takes_ge_5(ge_5)
+            takes_ge_5(ge_6)
+            # possibly should be allowed, but what if it's a float?
+            takes_ge_5(gt_4)  # E: incompatible_argument
+            takes_ge_5(gt_5)
+            takes_ge_5(4)  # E: incompatible_argument
+            takes_ge_5(5)
+            takes_ge_5(6)
+
+    @assert_passes()
+    def test_lt(self):
+        from typing_extensions import Annotated
+        from typing import Any
+        from annotated_types import Lt, Le
+
+        def takes_lt_5(x: Annotated[Any, Lt(5)]) -> None:
+            pass
+
+        def capybara(
+            i: int,
+            unnannotated,
+            lt_4: Annotated[int, Lt(4)],
+            lt_5: Annotated[int, Lt(5)],
+            lt_6: Annotated[int, Lt(6)],
+            le_4: Annotated[int, Le(4)],
+            le_5: Annotated[int, Le(5)],
+        ) -> None:
+            takes_lt_5(i)  # E: incompatible_argument
+            takes_lt_5(unnannotated)
+            takes_lt_5(lt_4)
+            takes_lt_5(lt_5)
+            takes_lt_5(lt_6)  # E: incompatible_argument
+            takes_lt_5(le_4)
+            takes_lt_5(le_5)  # E: incompatible_argument
+            takes_lt_5(4)
+            takes_lt_5(5)  # E: incompatible_argument
+
+    @assert_passes()
+    def test_le(self):
+        from typing_extensions import Annotated
+        from typing import Any
+        from annotated_types import Le, Lt
+
+        def takes_le_5(x: Annotated[Any, Le(5)]) -> None:
+            pass
+
+        def capybara(
+            i: int,
+            unnannotated,
+            le_4: Annotated[int, Le(4)],
+            le_5: Annotated[int, Le(5)],
+            le_6: Annotated[int, Le(6)],
+            lt_5: Annotated[int, Lt(5)],
+            lt_6: Annotated[int, Lt(6)],
+        ) -> None:
+            takes_le_5(i)  # E: incompatible_argument
+            takes_le_5(unnannotated)
+            takes_le_5(le_4)
+            takes_le_5(le_5)
+            takes_le_5(le_6)  # E: incompatible_argument
+            takes_le_5(lt_5)
+            # possibly should be allowed, but what if it's a float?
+            takes_le_5(lt_6)  # E: incompatible_argument
+            takes_le_5(4)
+            takes_le_5(5)
+            takes_le_5(6)  # E: incompatible_argument
+
+    @assert_passes()
+    def test_multiple_of(self):
+        from typing_extensions import Annotated
+        from annotated_types import MultipleOf
+
+        def takes_multiple_of_10(x: Annotated[int, MultipleOf(10)]) -> None:
+            pass
+
+        def capybara(
+            i: int,
+            unannotated,
+            multiple_of_2: Annotated[int, MultipleOf(2)],
+            multiple_of_10: Annotated[int, MultipleOf(10)],
+            multiple_of_20: Annotated[int, MultipleOf(20)],
+        ) -> None:
+            takes_multiple_of_10(i)  # E: incompatible_argument
+            takes_multiple_of_10(unannotated)
+            takes_multiple_of_10(multiple_of_2)  # E: incompatible_argument
+            takes_multiple_of_10(multiple_of_10)
+            takes_multiple_of_10(multiple_of_20)
+            takes_multiple_of_10(2)  # E: incompatible_argument
+            takes_multiple_of_10(10)
+            takes_multiple_of_10(50)
+
+    @assert_passes()
+    def test_min_max_len(self):
+        from typing_extensions import Annotated, TypedDict, NotRequired
+        from annotated_types import MinLen, MaxLen, Len
+
+        def takes_min_len_5(x: Annotated[object, MinLen(5)]) -> None:
+            pass
+
+        def takes_max_len_5(x: Annotated[object, MaxLen(5)]) -> None:
+            pass
+
+        def takes_3_to_5(x: Annotated[object, Len(3, 5)]) -> None:
+            pass
+
+        class TD3to5(TypedDict):
+            a: int
+            b: int
+            c: int
+            d: NotRequired[int]
+            e: NotRequired[int]
+
+        def capybara(
+            i: int,
+            min_4: Annotated[str, MinLen(4)],
+            min_6: Annotated[str, MinLen(6)],
+            max_4: Annotated[str, MaxLen(4)],
+            max_6: Annotated[str, MaxLen(6)],
+            exactly_5: Annotated[str, Len(5, 5)],
+            td: TD3to5,
+            unannotated,
+        ) -> None:
+            takes_min_len_5((1, 2, 3, 4))  # E: incompatible_argument
+            takes_min_len_5((1, 2, 3, 4, 5))
+            takes_min_len_5((1, 2, 3, 4, 5, 6))
+            takes_min_len_5((i, i, i, i))  # E: incompatible_argument
+            takes_min_len_5((i, i, i, i, i))
+            takes_min_len_5(min_4)  # E: incompatible_argument
+            takes_min_len_5(min_6)
+            takes_min_len_5(exactly_5)
+            takes_min_len_5(td)  # E: incompatible_argument
+            takes_min_len_5((i, i, i, i, i, *unannotated))
+
+            takes_max_len_5((1, 2, 3, 4, 5))
+            takes_max_len_5((1, 2, 3, 4, 5, 6))  # E: incompatible_argument
+            takes_max_len_5((i, i, i, i, i))
+            takes_max_len_5((i, i, i, i, i, i))  # E: incompatible_argument
+            takes_max_len_5(max_4)
+            takes_max_len_5(max_6)  # E: incompatible_argument
+            takes_max_len_5(exactly_5)
+            # TypedDict don't satisfy MaxLen as they're allowed to have
+            # arbitrary additional keys at runtime
+            takes_max_len_5(td)  # E: incompatible_argument
+            takes_max_len_5((i, *unannotated))  # E: incompatible_argument
+
+            takes_3_to_5((1, 2))  # E: incompatible_argument
+            takes_3_to_5((1, 2, 3))
+            takes_3_to_5(exactly_5)
+            takes_3_to_5(td)  # E: incompatible_argument
+
+    @assert_passes()
+    def test_timezone(self):
+        from typing_extensions import Annotated
+        from annotated_types import Timezone
+        from datetime import datetime, timezone, timedelta
+
+        def takes_naive(x: Annotated[datetime, Timezone(None)]) -> None:
+            pass
+
+        def takes_aware(x: Annotated[datetime, Timezone(...)]) -> None:
+            pass
+
+        def takes_utc(x: Annotated[datetime, Timezone(timezone.utc)]) -> None:
+            pass
+
+        naive_dt = datetime.now()
+        utc_dt = datetime.now(timezone.utc)
+        non_utc_aware_dt = datetime.now(timezone(timedelta(hours=1)))
+
+        def capybara(dt: datetime) -> None:
+            takes_naive(dt)  # E: incompatible_argument
+            takes_naive(naive_dt)
+            takes_naive(utc_dt)  # E: incompatible_argument
+            takes_naive(non_utc_aware_dt)  # E: incompatible_argument
+
+            takes_aware(dt)  # E: incompatible_argument
+            takes_aware(naive_dt)  # E: incompatible_argument
+            takes_aware(utc_dt)
+            takes_aware(non_utc_aware_dt)
+
+            takes_utc(dt)  # E: incompatible_argument
+            takes_utc(naive_dt)  # E: incompatible_argument
+            takes_utc(utc_dt)
+            takes_utc(non_utc_aware_dt)  # E: incompatible_argument
+
+    @assert_passes()
+    def test_predicate(self):
+        from typing_extensions import Annotated
+        from annotated_types import Predicate
+
+        def takes_upper(x: Annotated[str, Predicate(str.isupper)]) -> None:
+            pass
+
+        def capybara(
+            s: str, unannotated, scream: Annotated[str, Predicate(str.isupper)]
+        ) -> None:
+            takes_upper(s)  # E: incompatible_argument
+            takes_upper(unannotated)
+            takes_upper(scream)
+            takes_upper("WHY DO YOU ONLY WANT UPPERCASE")
+            takes_upper("lowercase")  # E: incompatible_argument
+
+
+class TestInferAnnotations(TestNameCheckVisitorBase):
+    @assert_passes()
+    def test_infer_gt(self):
+        from typing_extensions import Annotated
+        from annotated_types import Gt
+
+        def takes_gt_5(x: Annotated[int, Gt(5)]) -> None:
+            pass
+
+        def capybara(i: int) -> None:
+            takes_gt_5(i)  # E: incompatible_argument
+
+            if i > 4:
+                takes_gt_5(i)  # E: incompatible_argument
+
+            if i > 5:
+                takes_gt_5(i)
+
+            takes_gt_5(i)  # E: incompatible_argument
+
+            if i > 6:
+                takes_gt_5(i)
diff --git a/pyanalyze/test_stacked_scopes.py b/pyanalyze/test_stacked_scopes.py
index ca83e71..5e4aaa1 100644
--- a/pyanalyze/test_stacked_scopes.py
+++ b/pyanalyze/test_stacked_scopes.py
@@ -1493,6 +1493,31 @@ class TestConstraints(TestNameCheckVisitorBase):
                     ),
                 )
 
+    @assert_passes()
+    def test_hasattr_annotated_unification(self):
+        from pyanalyze.value import HasAttrExtension
+
+        ext = HasAttrExtension(KnownValue("name"), AnyValue(AnySource.inference))
+
+        def capybara(x: int) -> None:
+            assert_is_value(x, TypedValue(int))
+            if hasattr(x, "name"):
+                assert_is_value(x, AnnotatedValue(TypedValue(int), [ext]))
+            assert_is_value(x, TypedValue(int) | AnnotatedValue(TypedValue(int), [ext]))
+
+    @assert_passes()
+    def test_gt_annotated_unification(self):
+        from pyanalyze.value import CustomCheckExtension
+        from pyanalyze.annotated_types import Gt
+
+        ext = CustomCheckExtension(Gt(5))
+
+        def capybara(x: int) -> None:
+            assert_is_value(x, TypedValue(int))
+            if x > 5:
+                assert_is_value(x, AnnotatedValue(TypedValue(int), [ext]))
+            assert_is_value(x, TypedValue(int) | AnnotatedValue(TypedValue(int), [ext]))
+
     @assert_passes()
     def test_unconstrained_composite(self):
         class Foo(object):