add field_name_stack to get field name in __get_pydantic_core_schema__

pydantic/_internal/_annotated_handlers.py

@@ -109,6 +109,11 @@ def resolve_ref_schema(self, __maybe_ref_schema: core_schema.CoreSchema) -> core
         """
         raise NotImplementedError
 
+    @property
+    def field_name(self) -> str | None:
+        """Get the name of the closest field to this validator."""
+        raise NotImplementedError
+
     def _get_types_namespace(self) -> dict[str, Any] | None:
         """Internal method used during type resolution for serializer annotations."""
         raise NotImplementedError

pydantic/_internal/_generate_schema.py

@@ -262,6 +262,14 @@ def _add_custom_serialization_from_json_encoders(
 
 class GenerateSchema:
     """Generate core schema for a Pydantic model, dataclass and types like `str`, `datetime`, ... ."""
+    __slots__ = (
+        '_config_wrapper_stack',
+        '_types_namespace',
+        '_typevars_map',
+        '_needs_apply_discriminated_union',
+        'field_name_stack',
+        'defs',
+    )
 
     def __init__(
         self,
@@ -274,6 +282,7 @@ def __init__(
         self._types_namespace = types_namespace
         self._typevars_map = typevars_map
         self._needs_apply_discriminated_union = False
+        self.field_name_stack = _FieldNameStack()
         self.defs = _Definitions()
 
     @classmethod
@@ -289,6 +298,7 @@ def __from_parent(
         obj._types_namespace = types_namespace
         obj._typevars_map = typevars_map
         obj._needs_apply_discriminated_union = False
+        obj.field_name_stack = _FieldNameStack()
         obj.defs = defs
         return obj
 
@@ -883,13 +893,14 @@ def set_discriminator(schema: CoreSchema) -> CoreSchema:
             schema = self._apply_discriminator_to_union(schema, field_info.discriminator)
             return schema
 
-        if field_info.discriminator is not None:
-            schema = self._apply_annotations(source_type, annotations, transform_inner_schema=set_discriminator)
-        else:
-            schema = self._apply_annotations(
-                source_type,
-                annotations,
-            )
+        with self.field_name_stack.push(name):
+            if field_info.discriminator is not None:
+                schema = self._apply_annotations(source_type, annotations, transform_inner_schema=set_discriminator)
+            else:
+                schema = self._apply_annotations(
+                    source_type,
+                    annotations,
+                )
 
         # This V1 compatibility shim should eventually be removed
         # push down any `each_item=True` validators
@@ -1148,7 +1159,8 @@ def _generate_parameter_schema(
             field = FieldInfo.from_annotated_attribute(annotation, default)
         assert field.annotation is not None, 'field.annotation should not be None when generating a schema'
         source_type, annotations = field.annotation, field.metadata
-        schema = self._apply_annotations(source_type, annotations)
+        with self.field_name_stack.push(name):
+            schema = self._apply_annotations(source_type, annotations)
 
         if not field.is_required():
             schema = wrap_default(field, schema)
@@ -1980,3 +1992,22 @@ def resolve_original_schema(schema: CoreSchema, definitions: dict[str, CoreSchem
         return schema['schema']
     else:
         return schema
+
+
+class _FieldNameStack:
+    __slots__ = ('_stack',)
+
+    def __init__(self) -> None:
+        self._stack: list[str] = []
+
+    @contextmanager
+    def push(self, field_name: str) -> Iterator[None]:
+        self._stack.append(field_name)
+        yield
+        self._stack.pop()
+
+    def get(self) -> str | None:
+        if self._stack:
+            return self._stack[-1]
+        else:
+            return None

pydantic/_internal/_schema_generation_shared.py

@@ -95,6 +95,10 @@ def _get_types_namespace(self) -> dict[str, Any] | None:
     def generate_schema(self, __source_type: Any) -> core_schema.CoreSchema:
         return self._generate_schema.generate_schema(__source_type)
 
+    @property
+    def field_name(self) -> str | None:
+        return self._generate_schema.field_name_stack.get()
+
     def resolve_ref_schema(self, maybe_ref_schema: core_schema.CoreSchema) -> core_schema.CoreSchema:
         """Resolves reference in the core schema.