Skip to content

Schema Adapters

The schema adapter system provides a pluggable architecture for supporting different schema libraries in APIFlask.

Base Classes

SchemaAdapter

Bases: ABC

Base class for schema adapters.

Schema adapters provide a common interface for different schema libraries (marshmallow, Pydantic, etc.) to integrate with APIFlask.

Version added: 3.0.0

Source code in apiflask/schema_adapters/base.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
class SchemaAdapter(ABC):
    """Base class for schema adapters.

    Schema adapters provide a common interface for different schema libraries
    (marshmallow, Pydantic, etc.) to integrate with APIFlask.

    *Version added: 3.0.0*
    """

    def __init__(self, schema: t.Any, schema_name: str | None = None, many: bool = False) -> None:
        """Initialize the adapter with a schema.

        Arguments:
            schema: The schema object (marshmallow Schema, Pydantic model, etc.)
            schema_name: Optional schema name (used by some adapters like marshmallow)
            many: Whether this schema represents a list/array of items
        """
        self.schema = schema
        self.many = many

    @abstractmethod
    def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> t.Any:
        """Validate and parse input data from request.

        Arguments:
            request: Flask request object
            location: Location of data ('json', 'query', 'form', etc.)
            **kwargs: Additional arguments passed from decorator

        Returns:
            Validated and parsed data

        Raises:
            ValidationError: If validation fails
        """
        ...

    @abstractmethod
    def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
        """Serialize output data.

        Arguments:
            data: Data to serialize
            many: Whether to serialize many objects

        Returns:
            Serialized data ready for JSON response
        """
        ...

    @abstractmethod
    def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
        """Get OpenAPI schema definition.

        Arguments:
            **kwargs: Additional arguments for schema generation

        Returns:
            OpenAPI schema dict
        """
        ...

    @abstractmethod
    def get_schema_name(self) -> str:
        """Get the name of the schema for OpenAPI documentation.

        Returns:
            Schema name string
        """
        ...

    @property
    @abstractmethod
    def schema_type(self) -> str:
        """Get the type of schema adapter.

        Returns:
            Schema type identifier ('marshmallow', 'pydantic', etc.)
        """
        ...

schema_type abstractmethod property

Get the type of schema adapter.

Returns:

Type Description
str

Schema type identifier ('marshmallow', 'pydantic', etc.)

__init__(schema, schema_name=None, many=False)

Initialize the adapter with a schema.

Parameters:

Name Type Description Default
schema Any

The schema object (marshmallow Schema, Pydantic model, etc.)

 required
schema_name str | None

Optional schema name (used by some adapters like marshmallow)

 None
many bool

Whether this schema represents a list/array of items

 False
Source code in apiflask/schema_adapters/base.py 
20
21
22
23
24
25
26
27
28
29
def __init__(self, schema: t.Any, schema_name: str | None = None, many: bool = False) -> None:
    """Initialize the adapter with a schema.

    Arguments:
        schema: The schema object (marshmallow Schema, Pydantic model, etc.)
        schema_name: Optional schema name (used by some adapters like marshmallow)
        many: Whether this schema represents a list/array of items
    """
    self.schema = schema
    self.many = many

get_openapi_schema(**kwargs) abstractmethod

Get OpenAPI schema definition.

Parameters:

Name Type Description Default
**kwargs Any

Additional arguments for schema generation

 {}

Returns:

Type Description
dict[str, Any]

OpenAPI schema dict
Source code in apiflask/schema_adapters/base.py 
61
62
63
64
65
66
67
68
69
70
71
@abstractmethod
def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
    """Get OpenAPI schema definition.

    Arguments:
        **kwargs: Additional arguments for schema generation

    Returns:
        OpenAPI schema dict
    """
    ...

get_schema_name() abstractmethod

Get the name of the schema for OpenAPI documentation.

Returns:

Type Description
str

Schema name string
Source code in apiflask/schema_adapters/base.py 
73
74
75
76
77
78
79
80
@abstractmethod
def get_schema_name(self) -> str:
    """Get the name of the schema for OpenAPI documentation.

    Returns:
        Schema name string
    """
    ...

serialize_output(data, many=False) abstractmethod

Serialize output data.

Parameters:

Name Type Description Default
data Any

Data to serialize

 required
many bool

Whether to serialize many objects

 False

Returns:

Type Description
Any

Serialized data ready for JSON response
Source code in apiflask/schema_adapters/base.py 
48
49
50
51
52
53
54
55
56
57
58
59
@abstractmethod
def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
    """Serialize output data.

    Arguments:
        data: Data to serialize
        many: Whether to serialize many objects

    Returns:
        Serialized data ready for JSON response
    """
    ...

validate_input(request, location, **kwargs) abstractmethod

Validate and parse input data from request.

Parameters:

Name Type Description Default
request Request

Flask request object

 required
location str

Location of data ('json', 'query', 'form', etc.)

 required
**kwargs Any

Additional arguments passed from decorator

 {}

Returns:

Type Description
Any

Validated and parsed data

Raises:

Type Description
ValidationError

If validation fails
Source code in apiflask/schema_adapters/base.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
@abstractmethod
def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> t.Any:
    """Validate and parse input data from request.

    Arguments:
        request: Flask request object
        location: Location of data ('json', 'query', 'form', etc.)
        **kwargs: Additional arguments passed from decorator

    Returns:
        Validated and parsed data

    Raises:
        ValidationError: If validation fails
    """
    ...

Registry

SchemaRegistry

Registry for managing schema adapters and automatic type detection.

Version added: 3.0.0

Source code in apiflask/schema_adapters/registry.py 
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class SchemaRegistry:
    """Registry for managing schema adapters and automatic type detection.

    *Version added: 3.0.0*
    """

    def __init__(self) -> None:
        self._adapters: dict[str, type[SchemaAdapter]] = {}
        self._register_default_adapters()

    def _register_default_adapters(self) -> None:
        """Register the default schema adapters."""
        if HAS_MARSHMALLOW:
            from .marshmallow import MarshmallowAdapter

            self.register('marshmallow', MarshmallowAdapter)

        if HAS_PYDANTIC:
            from .pydantic import PydanticAdapter

            self.register('pydantic', PydanticAdapter)

    def register(self, name: str, adapter_class: type[SchemaAdapter]) -> None:
        """Register a schema adapter.

        Arguments:
            name: Name of the adapter
            adapter_class: Schema adapter class
        """
        self._adapters[name] = adapter_class

    def detect_schema_type(self, schema: t.Any) -> str:
        """Detect the type of schema.

        Arguments:
            schema: Schema object to detect type for

        Returns:
            Schema type name

        Raises:
            ValueError: If schema type cannot be detected
        """
        # Check for Pydantic models first
        if HAS_PYDANTIC:
            try:
                if isinstance(schema, type) and issubclass(schema, BaseModel):
                    return 'pydantic'
                elif isinstance(schema, BaseModel):
                    return 'pydantic'
            except TypeError:
                # Not a class that can be checked with issubclass
                pass

        # Check for marshmallow schemas
        if HAS_MARSHMALLOW:
            try:
                if isinstance(schema, Schema):
                    return 'marshmallow'
                elif isinstance(schema, type) and issubclass(schema, Schema):
                    return 'marshmallow'
                elif isinstance(schema, dict):  # Dict schemas are marshmallow
                    return 'marshmallow'
            except TypeError:
                # Not a class that can be checked with issubclass
                pass

        # Check for generic types like list[Model]
        if hasattr(schema, '__origin__') and hasattr(schema, '__args__'):
            # Handle generic types like list[Pet]
            if schema.__origin__ is list and len(schema.__args__) == 1:
                # Detect the inner type
                inner_type = schema.__args__[0]
                return self.detect_schema_type(inner_type)

        # Default to marshmallow for backwards compatibility
        if HAS_MARSHMALLOW:
            return 'marshmallow'

        raise ValueError(f'Cannot detect schema type for {type(schema)}')

    def create_adapter(
        self, schema: t.Any, schema_type: str | None = None, schema_name: str | None = None
    ) -> SchemaAdapter:
        """Create a schema adapter for the given schema.

        Arguments:
            schema: Schema object (can be a model or list[Model]/List[Model])
            schema_type: Optional schema type override
            schema_name: Optional schema name for dict schemas

        Returns:
            Schema adapter instance

        Raises:
            ValueError: If schema type is not supported
        """
        # Check if schema is a list type (list[Model] or List[Model])
        many = False
        inner_schema = schema

        if hasattr(schema, '__origin__') and hasattr(schema, '__args__'):
            if schema.__origin__ is list and len(schema.__args__) == 1:
                # Extract the inner model from list[Model] or List[Model]
                inner_schema = schema.__args__[0]
                many = True

        # Detect schema type from the inner schema
        if schema_type is None:
            schema_type = self.detect_schema_type(inner_schema)

        if schema_type not in self._adapters:
            available = ', '.join(self._adapters.keys())
            raise ValueError(
                f'Unsupported schema type: {schema_type}. ' f'Available types: {available}'
            )

        adapter_class = self._adapters[schema_type]
        return adapter_class(inner_schema, schema_name=schema_name, many=many)

    def get_available_types(self) -> list[str]:
        """Get list of available schema types.

        Returns:
            List of available schema type names
        """
        return list(self._adapters.keys())

create_adapter(schema, schema_type=None, schema_name=None)

Create a schema adapter for the given schema.

Parameters:

Name Type Description Default
schema Any

Schema object (can be a model or list[Model]/List[Model])

 required
schema_type str | None

Optional schema type override

 None
schema_name str | None

Optional schema name for dict schemas

 None

Returns:

Type Description
SchemaAdapter

Schema adapter instance

Raises:

Type Description
ValueError

If schema type is not supported
Source code in apiflask/schema_adapters/registry.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def create_adapter(
    self, schema: t.Any, schema_type: str | None = None, schema_name: str | None = None
) -> SchemaAdapter:
    """Create a schema adapter for the given schema.

    Arguments:
        schema: Schema object (can be a model or list[Model]/List[Model])
        schema_type: Optional schema type override
        schema_name: Optional schema name for dict schemas

    Returns:
        Schema adapter instance

    Raises:
        ValueError: If schema type is not supported
    """
    # Check if schema is a list type (list[Model] or List[Model])
    many = False
    inner_schema = schema

    if hasattr(schema, '__origin__') and hasattr(schema, '__args__'):
        if schema.__origin__ is list and len(schema.__args__) == 1:
            # Extract the inner model from list[Model] or List[Model]
            inner_schema = schema.__args__[0]
            many = True

    # Detect schema type from the inner schema
    if schema_type is None:
        schema_type = self.detect_schema_type(inner_schema)

    if schema_type not in self._adapters:
        available = ', '.join(self._adapters.keys())
        raise ValueError(
            f'Unsupported schema type: {schema_type}. ' f'Available types: {available}'
        )

    adapter_class = self._adapters[schema_type]
    return adapter_class(inner_schema, schema_name=schema_name, many=many)

detect_schema_type(schema)

Detect the type of schema.

Parameters:

Name Type Description Default
schema Any

Schema object to detect type for

 required

Returns:

Type Description
str

Schema type name

Raises:

Type Description
ValueError

If schema type cannot be detected
Source code in apiflask/schema_adapters/registry.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def detect_schema_type(self, schema: t.Any) -> str:
    """Detect the type of schema.

    Arguments:
        schema: Schema object to detect type for

    Returns:
        Schema type name

    Raises:
        ValueError: If schema type cannot be detected
    """
    # Check for Pydantic models first
    if HAS_PYDANTIC:
        try:
            if isinstance(schema, type) and issubclass(schema, BaseModel):
                return 'pydantic'
            elif isinstance(schema, BaseModel):
                return 'pydantic'
        except TypeError:
            # Not a class that can be checked with issubclass
            pass

    # Check for marshmallow schemas
    if HAS_MARSHMALLOW:
        try:
            if isinstance(schema, Schema):
                return 'marshmallow'
            elif isinstance(schema, type) and issubclass(schema, Schema):
                return 'marshmallow'
            elif isinstance(schema, dict):  # Dict schemas are marshmallow
                return 'marshmallow'
        except TypeError:
            # Not a class that can be checked with issubclass
            pass

    # Check for generic types like list[Model]
    if hasattr(schema, '__origin__') and hasattr(schema, '__args__'):
        # Handle generic types like list[Pet]
        if schema.__origin__ is list and len(schema.__args__) == 1:
            # Detect the inner type
            inner_type = schema.__args__[0]
            return self.detect_schema_type(inner_type)

    # Default to marshmallow for backwards compatibility
    if HAS_MARSHMALLOW:
        return 'marshmallow'

    raise ValueError(f'Cannot detect schema type for {type(schema)}')

get_available_types()

Get list of available schema types.

Returns:

Type Description
list[str]

List of available schema type names
Source code in apiflask/schema_adapters/registry.py 
147
148
149
150
151
152
153
def get_available_types(self) -> list[str]:
    """Get list of available schema types.

    Returns:
        List of available schema type names
    """
    return list(self._adapters.keys())

register(name, adapter_class)

Register a schema adapter.

Parameters:

Name Type Description Default
name str

Name of the adapter

 required
adapter_class type[SchemaAdapter]

Schema adapter class

 required
Source code in apiflask/schema_adapters/registry.py 
49
50
51
52
53
54
55
56
def register(self, name: str, adapter_class: type[SchemaAdapter]) -> None:
    """Register a schema adapter.

    Arguments:
        name: Name of the adapter
        adapter_class: Schema adapter class
    """
    self._adapters[name] = adapter_class

Adapters

marshmallow Adapter

FlaskParser

Bases: FlaskParser

Overwrite the default webargs.FlaskParser.handle_error.

Update the default status code and the error description from related configuration variables.

Version added: 3.0.0

Source code in apiflask/schema_adapters/marshmallow.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
class FlaskParser(BaseFlaskParser):
    """Overwrite the default `webargs.FlaskParser.handle_error`.

    Update the default status code and the error description from related
    configuration variables.

    *Version added: 3.0.0*
    """

    USE_ARGS_POSITIONAL = False

    def handle_error(  # type: ignore
        self,
        error: MarshmallowValidationError,
        req: Request,
        schema: Schema,
        *,
        error_status_code: int,
        error_headers: t.Mapping[str, str],
    ) -> None:
        raise _ValidationError(
            error_status_code or current_app.config['VALIDATION_ERROR_STATUS_CODE'],
            current_app.config['VALIDATION_ERROR_DESCRIPTION'],
            error.messages,
            error_headers,
        )

    def load_location_data(self, *, schema: Schema, req: Request, location: str) -> t.Any:
        return self._load_location_data(schema=schema, req=req, location=location)

MarshmallowAdapter

Bases: SchemaAdapter

Schema adapter for marshmallow schemas.

Source code in apiflask/schema_adapters/marshmallow.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
class MarshmallowAdapter(SchemaAdapter):
    """Schema adapter for marshmallow schemas."""

    def __init__(
        self,
        schema: Schema | dict | type[Schema],
        schema_name: str | None = None,
        many: bool = False,
    ) -> None:
        """Initialize the marshmallow adapter.

        Arguments:
            schema: Marshmallow schema instance, dict, or schema class
            schema_name: Optional schema name for dict schemas
            many: Whether this schema represents a list/array of items
        """
        if isinstance(schema, dict):
            # Convert dict schema to marshmallow schema
            if schema_name is None:
                schema_name = 'GeneratedSchema'
            if schema == {}:
                self.schema = EmptySchema()
            else:
                self.schema = Schema.from_dict(schema, name=schema_name)()  # type: ignore
        elif isinstance(schema, type):
            # Check if it's a marshmallow schema class
            try:
                if issubclass(schema, Schema):
                    self.schema = schema()
                else:
                    # If not a marshmallow schema, keep as class (might be Pydantic)
                    self.schema = schema  # type: ignore[unreachable]
            except TypeError:
                # Not a class that can be checked with issubclass
                self.schema = schema
        else:
            self.schema = schema

        self.many = many

    @property
    def schema_type(self) -> str:
        return 'marshmallow'

    def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> t.Any:
        """Validate input using marshmallow/webargs."""
        if location == 'files':
            # Handle file uploads with form data
            data = _get_files_and_form(request, self.schema)
            try:
                return self.schema.load(data)
            except MarshmallowValidationError as error:
                raise _ValidationError(
                    current_app.config['VALIDATION_ERROR_STATUS_CODE'],
                    current_app.config['VALIDATION_ERROR_DESCRIPTION'],
                    error.messages,
                ) from error

        # Use webargs for other locations
        return parser.load_location_data(schema=self.schema, req=request, location=location)

    def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
        """Serialize output using marshmallow."""
        if isinstance(self.schema, (EmptySchema, FileSchema)):
            return data

        # Use marshmallow's dump method which handles dump_default values
        return self.schema.dump(data, many=many)

    def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
        """Get OpenAPI schema from marshmallow schema."""
        if isinstance(self.schema, EmptySchema):
            return {}
        elif isinstance(self.schema, FileSchema):
            return {'type': self.schema.type, 'format': self.schema.format}

        from apispec.ext.marshmallow import MarshmallowPlugin

        # MarshmallowPlugin expects schema_name_resolver to be None or callable[[type[Schema]], str]
        # Since we don't need custom resolution here, pass None
        plugin = MarshmallowPlugin(schema_name_resolver=None)
        result: dict[str, t.Any] = plugin.schema_helper(self.schema, **kwargs)  # type: ignore[no-untyped-call]
        return result

    def get_schema_name(self) -> str:
        """Get schema name for OpenAPI documentation.

        *Version changed: 3.0.0*

        - Remove "Schema" suffix stripping from schema names.

        """
        schema = self.schema
        if isinstance(schema, type):  # pragma: no cover
            schema = schema()  # type: ignore

        name: str = schema.__class__.__name__
        if hasattr(schema, 'partial') and schema.partial:
            name += 'Update'
        return name

__init__(schema, schema_name=None, many=False)

Initialize the marshmallow adapter.

Parameters:

Name Type Description Default
schema Schema | dict | type[Schema]

Marshmallow schema instance, dict, or schema class

 required
schema_name str | None

Optional schema name for dict schemas

 None
many bool

Whether this schema represents a list/array of items

 False
Source code in apiflask/schema_adapters/marshmallow.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def __init__(
    self,
    schema: Schema | dict | type[Schema],
    schema_name: str | None = None,
    many: bool = False,
) -> None:
    """Initialize the marshmallow adapter.

    Arguments:
        schema: Marshmallow schema instance, dict, or schema class
        schema_name: Optional schema name for dict schemas
        many: Whether this schema represents a list/array of items
    """
    if isinstance(schema, dict):
        # Convert dict schema to marshmallow schema
        if schema_name is None:
            schema_name = 'GeneratedSchema'
        if schema == {}:
            self.schema = EmptySchema()
        else:
            self.schema = Schema.from_dict(schema, name=schema_name)()  # type: ignore
    elif isinstance(schema, type):
        # Check if it's a marshmallow schema class
        try:
            if issubclass(schema, Schema):
                self.schema = schema()
            else:
                # If not a marshmallow schema, keep as class (might be Pydantic)
                self.schema = schema  # type: ignore[unreachable]
        except TypeError:
            # Not a class that can be checked with issubclass
            self.schema = schema
    else:
        self.schema = schema

    self.many = many

get_openapi_schema(**kwargs)

Get OpenAPI schema from marshmallow schema.

Source code in apiflask/schema_adapters/marshmallow.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
    """Get OpenAPI schema from marshmallow schema."""
    if isinstance(self.schema, EmptySchema):
        return {}
    elif isinstance(self.schema, FileSchema):
        return {'type': self.schema.type, 'format': self.schema.format}

    from apispec.ext.marshmallow import MarshmallowPlugin

    # MarshmallowPlugin expects schema_name_resolver to be None or callable[[type[Schema]], str]
    # Since we don't need custom resolution here, pass None
    plugin = MarshmallowPlugin(schema_name_resolver=None)
    result: dict[str, t.Any] = plugin.schema_helper(self.schema, **kwargs)  # type: ignore[no-untyped-call]
    return result

get_schema_name()

Get schema name for OpenAPI documentation.

Version changed: 3.0.0

  • Remove "Schema" suffix stripping from schema names.
Source code in apiflask/schema_adapters/marshmallow.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_schema_name(self) -> str:
    """Get schema name for OpenAPI documentation.

    *Version changed: 3.0.0*

    - Remove "Schema" suffix stripping from schema names.

    """
    schema = self.schema
    if isinstance(schema, type):  # pragma: no cover
        schema = schema()  # type: ignore

    name: str = schema.__class__.__name__
    if hasattr(schema, 'partial') and schema.partial:
        name += 'Update'
    return name

serialize_output(data, many=False)

Serialize output using marshmallow.

Source code in apiflask/schema_adapters/marshmallow.py 
136
137
138
139
140
141
142
def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
    """Serialize output using marshmallow."""
    if isinstance(self.schema, (EmptySchema, FileSchema)):
        return data

    # Use marshmallow's dump method which handles dump_default values
    return self.schema.dump(data, many=many)

validate_input(request, location, **kwargs)

Validate input using marshmallow/webargs.

Source code in apiflask/schema_adapters/marshmallow.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> t.Any:
    """Validate input using marshmallow/webargs."""
    if location == 'files':
        # Handle file uploads with form data
        data = _get_files_and_form(request, self.schema)
        try:
            return self.schema.load(data)
        except MarshmallowValidationError as error:
            raise _ValidationError(
                current_app.config['VALIDATION_ERROR_STATUS_CODE'],
                current_app.config['VALIDATION_ERROR_DESCRIPTION'],
                error.messages,
            ) from error

    # Use webargs for other locations
    return parser.load_location_data(schema=self.schema, req=request, location=location)

Pydantic Adapter

PydanticAdapter

Bases: SchemaAdapter

Schema adapter for Pydantic models.

Source code in apiflask/schema_adapters/pydantic.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
class PydanticAdapter(SchemaAdapter):
    """Schema adapter for Pydantic models."""

    def __init__(
        self,
        schema: type[BaseModel] | BaseModel,
        schema_name: str | None = None,
        many: bool = False,
    ) -> None:
        """Initialize the Pydantic adapter.

        Arguments:
            schema: Pydantic model class or instance
            schema_name: Optional schema name (not used for Pydantic)
            many: Whether this schema represents a list/array of items
        """
        if not HAS_PYDANTIC:
            raise ImportError(
                'Pydantic is required for PydanticAdapter. ' 'Install it with: pip install pydantic'
            )

        if isinstance(schema, type) and issubclass(schema, BaseModel):
            self.model_class = schema
            self.schema = schema
        elif isinstance(schema, BaseModel):
            self.model_class = schema.__class__
            self.schema = schema.__class__
        else:
            raise TypeError(f'Expected Pydantic model, got {type(schema)}')

        self.many = many

    @property
    def schema_type(self) -> str:
        return 'pydantic'

    def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> BaseModel:
        """Validate input using Pydantic."""
        try:
            if location == 'json':
                data = request.get_json(force=True)
                if data is None:
                    data = {}
                return self.model_class.model_validate(data)

            elif location == 'query' or location == 'querystring':
                # Handle query parameters
                data = request.args.to_dict()
                return self.model_class.model_validate(data)

            elif location == 'form':
                # Handle form data
                data = request.form.to_dict()
                return self.model_class.model_validate(data)

            elif location == 'files':
                # Handle file uploads with form data
                data = {}
                data.update(request.form.to_dict())

                # Add files to data
                for key, file in request.files.items():
                    if isinstance(file, IOBase):
                        data[key] = file

                return self.model_class.model_validate(data)

            elif location == 'form_and_files':
                # Combine form and files
                data = request.form.to_dict()
                for key, file in request.files.items():
                    data[key] = file  # type: ignore
                return self.model_class.model_validate(data)

            elif location == 'json_or_form':
                # Try JSON first, then form
                if request.is_json:
                    data = request.get_json(force=True) or {}
                else:
                    data = request.form.to_dict()
                return self.model_class.model_validate(data)

            elif location == 'cookies':
                # Handle cookies
                data = request.cookies.to_dict()
                return self.model_class.model_validate(data)

            elif location == 'headers':
                # Handle headers - convert header names to field names
                # HTTP headers like X-Token become x_token for Pydantic fields
                data = {}
                for header_name, value in request.headers:
                    # Convert header name to field name (e.g., X-Token -> x_token)
                    field_name = header_name.lower().replace('-', '_')
                    data[field_name] = value
                return self.model_class.model_validate(data)

            elif location == 'path' or location == 'view_args':
                # Handle path/view_args
                data = request.view_args or {}
                return self.model_class.model_validate(data)

            else:
                raise ValueError(f'Unsupported location: {location}')

        except PydanticValidationError as error:
            formatted_errors = _format_pydantic_errors(error.errors())
            raise _ValidationError(
                current_app.config['VALIDATION_ERROR_STATUS_CODE'],
                current_app.config['VALIDATION_ERROR_DESCRIPTION'],
                {location: formatted_errors},
            ) from error

    def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
        """Serialize output using Pydantic with validation."""
        if many and isinstance(data, (list, tuple)):
            # Handle lists of data
            result = []
            for item in data:
                if isinstance(item, BaseModel):
                    # Already validated, just serialize
                    result.append(item.model_dump(mode='json', by_alias=True))
                else:
                    # Validate and serialize
                    validated = self.model_class.model_validate(item)
                    result.append(validated.model_dump(mode='json', by_alias=True))
            return result
        elif isinstance(data, BaseModel):
            # Pydantic model instance - already validated, just serialize
            return data.model_dump(mode='json', by_alias=True)
        elif isinstance(data, (list, tuple)) and not many:
            # Handle lists when many=False
            result = []
            for item in data:
                if isinstance(item, BaseModel):
                    result.append(item.model_dump(mode='json', by_alias=True))
                else:
                    # Validate and serialize
                    validated = self.model_class.model_validate(item)
                    result.append(validated.model_dump(mode='json', by_alias=True))
            return result
        else:
            # Validate and serialize (dicts, primitives)
            validated = self.model_class.model_validate(data)
            return validated.model_dump(mode='json', by_alias=True)

    def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
        """Get OpenAPI schema from Pydantic model.

        Uses Pydantic's ref_template to generate OpenAPI-compliant $ref paths.
        Nested models in $defs should be extracted and registered separately
        at the components/schemas level.
        """
        try:
            # Generate schema name for use in ref_template
            schema_name = self.get_schema_name()

            # Use ref_template to make Pydantic generate OpenAPI-compatible refs
            # The {model} placeholder will be replaced with nested model names
            ref_template = f'#/components/schemas/{schema_name}.{{model}}'

            # Generate JSON schema with OpenAPI-compatible references
            schema = self.model_class.model_json_schema(
                ref_template=ref_template, mode=kwargs.get('mode', 'validation')
            )

            # Note: The $defs should be extracted and registered separately
            # in components/schemas. The caller (app.py) should handle this.

            return schema

        except Exception:
            # Fallback to basic schema
            return {'type': 'object', 'title': self.get_schema_name()}

    def get_schema_name(self) -> str:
        """Get schema name for OpenAPI documentation."""
        return self.model_class.__name__

__init__(schema, schema_name=None, many=False)

Initialize the Pydantic adapter.

Parameters:

Name Type Description Default
schema type[BaseModel] | BaseModel

Pydantic model class or instance

 required
schema_name str | None

Optional schema name (not used for Pydantic)

 None
many bool

Whether this schema represents a list/array of items

 False
Source code in apiflask/schema_adapters/pydantic.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def __init__(
    self,
    schema: type[BaseModel] | BaseModel,
    schema_name: str | None = None,
    many: bool = False,
) -> None:
    """Initialize the Pydantic adapter.

    Arguments:
        schema: Pydantic model class or instance
        schema_name: Optional schema name (not used for Pydantic)
        many: Whether this schema represents a list/array of items
    """
    if not HAS_PYDANTIC:
        raise ImportError(
            'Pydantic is required for PydanticAdapter. ' 'Install it with: pip install pydantic'
        )

    if isinstance(schema, type) and issubclass(schema, BaseModel):
        self.model_class = schema
        self.schema = schema
    elif isinstance(schema, BaseModel):
        self.model_class = schema.__class__
        self.schema = schema.__class__
    else:
        raise TypeError(f'Expected Pydantic model, got {type(schema)}')

    self.many = many

get_openapi_schema(**kwargs)

Get OpenAPI schema from Pydantic model.

Uses Pydantic's ref_template to generate OpenAPI-compliant $ref paths. Nested models in $defs should be extracted and registered separately at the components/schemas level.

Source code in apiflask/schema_adapters/pydantic.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def get_openapi_schema(self, **kwargs: t.Any) -> dict[str, t.Any]:
    """Get OpenAPI schema from Pydantic model.

    Uses Pydantic's ref_template to generate OpenAPI-compliant $ref paths.
    Nested models in $defs should be extracted and registered separately
    at the components/schemas level.
    """
    try:
        # Generate schema name for use in ref_template
        schema_name = self.get_schema_name()

        # Use ref_template to make Pydantic generate OpenAPI-compatible refs
        # The {model} placeholder will be replaced with nested model names
        ref_template = f'#/components/schemas/{schema_name}.{{model}}'

        # Generate JSON schema with OpenAPI-compatible references
        schema = self.model_class.model_json_schema(
            ref_template=ref_template, mode=kwargs.get('mode', 'validation')
        )

        # Note: The $defs should be extracted and registered separately
        # in components/schemas. The caller (app.py) should handle this.

        return schema

    except Exception:
        # Fallback to basic schema
        return {'type': 'object', 'title': self.get_schema_name()}

get_schema_name()

Get schema name for OpenAPI documentation.

Source code in apiflask/schema_adapters/pydantic.py 
230
231
232
def get_schema_name(self) -> str:
    """Get schema name for OpenAPI documentation."""
    return self.model_class.__name__

serialize_output(data, many=False)

Serialize output using Pydantic with validation.

Source code in apiflask/schema_adapters/pydantic.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
def serialize_output(self, data: t.Any, many: bool = False) -> t.Any:
    """Serialize output using Pydantic with validation."""
    if many and isinstance(data, (list, tuple)):
        # Handle lists of data
        result = []
        for item in data:
            if isinstance(item, BaseModel):
                # Already validated, just serialize
                result.append(item.model_dump(mode='json', by_alias=True))
            else:
                # Validate and serialize
                validated = self.model_class.model_validate(item)
                result.append(validated.model_dump(mode='json', by_alias=True))
        return result
    elif isinstance(data, BaseModel):
        # Pydantic model instance - already validated, just serialize
        return data.model_dump(mode='json', by_alias=True)
    elif isinstance(data, (list, tuple)) and not many:
        # Handle lists when many=False
        result = []
        for item in data:
            if isinstance(item, BaseModel):
                result.append(item.model_dump(mode='json', by_alias=True))
            else:
                # Validate and serialize
                validated = self.model_class.model_validate(item)
                result.append(validated.model_dump(mode='json', by_alias=True))
        return result
    else:
        # Validate and serialize (dicts, primitives)
        validated = self.model_class.model_validate(data)
        return validated.model_dump(mode='json', by_alias=True)

validate_input(request, location, **kwargs)

Validate input using Pydantic.

Source code in apiflask/schema_adapters/pydantic.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def validate_input(self, request: Request, location: str, **kwargs: t.Any) -> BaseModel:
    """Validate input using Pydantic."""
    try:
        if location == 'json':
            data = request.get_json(force=True)
            if data is None:
                data = {}
            return self.model_class.model_validate(data)

        elif location == 'query' or location == 'querystring':
            # Handle query parameters
            data = request.args.to_dict()
            return self.model_class.model_validate(data)

        elif location == 'form':
            # Handle form data
            data = request.form.to_dict()
            return self.model_class.model_validate(data)

        elif location == 'files':
            # Handle file uploads with form data
            data = {}
            data.update(request.form.to_dict())

            # Add files to data
            for key, file in request.files.items():
                if isinstance(file, IOBase):
                    data[key] = file

            return self.model_class.model_validate(data)

        elif location == 'form_and_files':
            # Combine form and files
            data = request.form.to_dict()
            for key, file in request.files.items():
                data[key] = file  # type: ignore
            return self.model_class.model_validate(data)

        elif location == 'json_or_form':
            # Try JSON first, then form
            if request.is_json:
                data = request.get_json(force=True) or {}
            else:
                data = request.form.to_dict()
            return self.model_class.model_validate(data)

        elif location == 'cookies':
            # Handle cookies
            data = request.cookies.to_dict()
            return self.model_class.model_validate(data)

        elif location == 'headers':
            # Handle headers - convert header names to field names
            # HTTP headers like X-Token become x_token for Pydantic fields
            data = {}
            for header_name, value in request.headers:
                # Convert header name to field name (e.g., X-Token -> x_token)
                field_name = header_name.lower().replace('-', '_')
                data[field_name] = value
            return self.model_class.model_validate(data)

        elif location == 'path' or location == 'view_args':
            # Handle path/view_args
            data = request.view_args or {}
            return self.model_class.model_validate(data)

        else:
            raise ValueError(f'Unsupported location: {location}')

    except PydanticValidationError as error:
        formatted_errors = _format_pydantic_errors(error.errors())
        raise _ValidationError(
            current_app.config['VALIDATION_ERROR_STATUS_CODE'],
            current_app.config['VALIDATION_ERROR_DESCRIPTION'],
            {location: formatted_errors},
        ) from error