SQLAlchemy integration with the marshmallow (de)serialization library
—
Comprehensive system for converting SQLAlchemy columns, properties, and models into appropriate marshmallow fields, with support for all standard SQLAlchemy types and database-specific extensions.
Note: Type annotations using Literal require from typing import Literal (Python 3.8+) or from typing_extensions import Literal (earlier versions).
The core conversion class that translates SQLAlchemy constructs to marshmallow fields with comprehensive type mapping and customization options.
class ModelConverter:
"""Converts SQLAlchemy models into marshmallow fields.
Provides methods for converting entire models, individual properties,
and columns to appropriate marshmallow field types.
"""
def __init__(self, schema_cls: type[ma.Schema] | None = None): ...
def fields_for_model(
self,
model: type[DeclarativeMeta],
*,
include_fk: bool = False,
include_relationships: bool = False,
fields: Iterable[str] | None = None,
exclude: Iterable[str] | None = None,
base_fields: dict | None = None,
dict_cls: type[dict] = dict,
) -> dict[str, fields.Field]:
"""Generate dict of field_name: Field pairs for the given model.
Parameters:
- model: SQLAlchemy model class
- include_fk: Whether to include foreign key fields
- include_relationships: Whether to include relationship fields
- fields: Only include these field names (whitelist)
- exclude: Exclude these field names (blacklist)
- base_fields: Existing fields to merge with
- dict_cls: Dict class to use for result
Returns:
Dictionary mapping field names to marshmallow Field instances
"""
def fields_for_table(
self,
table: sa.Table,
*,
include_fk: bool = False,
fields: Iterable[str] | None = None,
exclude: Iterable[str] | None = None,
base_fields: dict | None = None,
dict_cls: type[dict] = dict,
) -> dict[str, fields.Field]:
"""Generate dict of field_name: Field pairs for the given table.
Parameters:
- table: SQLAlchemy Table instance
- include_fk: Whether to include foreign key fields
- fields: Only include these field names (whitelist)
- exclude: Exclude these field names (blacklist)
- base_fields: Existing fields to merge with
- dict_cls: Dict class to use for result
Returns:
Dictionary mapping field names to marshmallow Field instances
"""
@overload
def property2field(
self,
prop: MapperProperty,
*,
instance: Literal[True] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> fields.Field: ...
@overload
def property2field(
self,
prop: MapperProperty,
*,
instance: Literal[False] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> type[fields.Field]: ...
def property2field(
self,
prop: MapperProperty,
*,
instance: bool = True,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> fields.Field | type[fields.Field]:
"""Convert SQLAlchemy Property to marshmallow field.
Parameters:
- prop: SQLAlchemy Property (column property or relationship)
- instance: If True, return Field instance; if False, return Field class
- field_class: Override field class to use
- **kwargs: Additional field constructor arguments
Returns:
Field instance or class depending on instance parameter
"""
@overload
def column2field(
self,
column: sa.Column,
*,
instance: Literal[True] = ...,
**kwargs
) -> fields.Field: ...
@overload
def column2field(
self,
column: sa.Column,
*,
instance: Literal[False] = ...,
**kwargs
) -> type[fields.Field]: ...
def column2field(
self,
column: sa.Column,
*,
instance: bool = True,
**kwargs
) -> fields.Field | type[fields.Field]:
"""Convert SQLAlchemy Column to marshmallow field.
Parameters:
- column: SQLAlchemy Column instance
- instance: If True, return Field instance; if False, return Field class
- **kwargs: Additional field constructor arguments
Returns:
Field instance or class depending on instance parameter
"""
@overload
def field_for(
self,
model: type[DeclarativeMeta],
property_name: str,
*,
instance: Literal[True] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> fields.Field: ...
@overload
def field_for(
self,
model: type[DeclarativeMeta],
property_name: str,
*,
instance: Literal[False] = ...,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> type[fields.Field]: ...
def field_for(
self,
model: type[DeclarativeMeta],
property_name: str,
*,
instance: bool = True,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> fields.Field | type[fields.Field]:
"""Convert model property to marshmallow field by name.
Parameters:
- model: SQLAlchemy model class
- property_name: Name of property to convert
- instance: If True, return Field instance; if False, return Field class
- field_class: Override field class to use
- **kwargs: Additional field constructor arguments
Returns:
Field instance or class depending on instance parameter
"""Module-level convenience functions that use a default ModelConverter instance for quick field conversion.
def fields_for_model(
model: type[DeclarativeMeta],
*,
include_fk: bool = False,
include_relationships: bool = False,
fields: Iterable[str] | None = None,
exclude: Iterable[str] | None = None,
base_fields: dict | None = None,
dict_cls: type[dict] = dict,
) -> dict[str, fields.Field]:
"""Generate fields dict from SQLAlchemy model (convenience function).
Uses default ModelConverter instance.
"""
@overload
def property2field(
prop: MapperProperty,
*,
instance: Literal[True] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> fields.Field: ...
@overload
def property2field(
prop: MapperProperty,
*,
instance: Literal[False] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> type[fields.Field]: ...
def property2field(
prop: MapperProperty,
*,
instance: bool = True,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> fields.Field | type[fields.Field]:
"""Convert SQLAlchemy property to field (convenience function).
Uses default ModelConverter instance.
"""
@overload
def column2field(
column: sa.Column,
*,
instance: Literal[True] = ...,
**kwargs
) -> fields.Field: ...
@overload
def column2field(
column: sa.Column,
*,
instance: Literal[False] = ...,
**kwargs
) -> type[fields.Field]: ...
def column2field(
column: sa.Column,
*,
instance: bool = True,
**kwargs
) -> fields.Field | type[fields.Field]:
"""Convert SQLAlchemy column to field (convenience function).
Uses default ModelConverter instance.
"""
@overload
def field_for(
model: type[DeclarativeMeta],
property_name: str,
*,
instance: Literal[True] = ...,
field_class: type[fields.Field] | None = ...,
**kwargs,
) -> fields.Field: ...
@overload
def field_for(
model: type[DeclarativeMeta],
property_name: str,
*,
instance: Literal[False] = ...,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> type[fields.Field]: ...
def field_for(
model: type[DeclarativeMeta],
property_name: str,
*,
instance: bool = True,
field_class: type[fields.Field] | None = None,
**kwargs,
) -> fields.Field | type[fields.Field]:
"""Get field for model property by name (convenience function).
Uses default ModelConverter instance.
"""from marshmallow_sqlalchemy import fields_for_model, ModelConverter
from mymodels import User, Post
# Using convenience function
user_fields = fields_for_model(User)
# Returns: {"id": Integer(), "name": String(), "email": String(), ...}
# Include relationships and foreign keys
post_fields = fields_for_model(
Post,
include_fk=True,
include_relationships=True
)
# Exclude specific fields
user_fields_filtered = fields_for_model(
User,
exclude=["password_hash", "last_login"]
)
# Only include specific fields
user_fields_minimal = fields_for_model(
User,
fields=["id", "name", "email"]
)from marshmallow_sqlalchemy import field_for, property2field
from mymodels import User
# Get field for specific property
name_field = field_for(User, "name")
email_field = field_for(User, "email", dump_only=True)
# Get field class instead of instance
NameFieldClass = field_for(User, "name", instance=False)
# Convert property directly
user_mapper = sa.inspect(User)
name_prop = user_mapper.attrs["name"]
name_field = property2field(name_prop)from marshmallow_sqlalchemy import column2field
from mymodels import User
# Convert table column
user_table = User.__table__
name_column = user_table.columns["name"]
name_field = column2field(name_column)
# Add custom validation
email_field = column2field(
user_table.columns["email"],
validate=[validate.Email(), validate.Length(max=255)]
)from marshmallow_sqlalchemy import ModelConverter
from marshmallow import fields
class CustomConverter(ModelConverter):
"""Custom converter with additional type mappings."""
SQLA_TYPE_MAPPING = {
**ModelConverter.SQLA_TYPE_MAPPING,
MyCustomType: fields.String, # Map custom type to String field
}
def _get_field_class_for_data_type(self, data_type):
# Custom logic for field type selection
if isinstance(data_type, MySpecialType):
return fields.Raw
return super()._get_field_class_for_data_type(data_type)
# Use custom converter
converter = CustomConverter()
fields_dict = converter.fields_for_model(MyModel)
# Or use with schema
class MySchema(SQLAlchemyAutoSchema):
class Meta:
model = MyModel
model_converter = CustomConverterModelConverter includes comprehensive mappings for SQLAlchemy types:
sa.String → fields.Stringsa.Integer → fields.Integersa.Float → fields.Floatsa.Boolean → fields.Booleansa.DateTime → fields.DateTimesa.Date → fields.Datesa.Time → fields.Timesa.Numeric → fields.Decimalsa.Text → fields.Stringsa.JSON → fields.Rawsa.Enum → fields.Enum (if enum_class) or fields.Rawsa.PickleType → fields.Rawpostgresql.UUID → fields.UUIDpostgresql.JSONB → fields.Rawpostgresql.ARRAY → fields.Listpostgresql.HSTORE → fields.Rawpostgresql.INET → fields.Stringpostgresql.CIDR → fields.Stringpostgresql.MACADDR → fields.Stringpostgresql.MONEY → fields.Decimalmysql.BIT → fields.Integermysql.YEAR → fields.Integermysql.SET → fields.Listmysql.ENUM → fields.Fieldmssql.BIT → fields.Integermssql.UNIQUEIDENTIFIER → fields.UUIDThe converter automatically configures fields based on column properties:
# Column with nullable=False becomes required field
name = sa.Column(sa.String, nullable=False)
# → fields.String(required=True)
# Column with default value becomes non-required
status = sa.Column(sa.String, default="active")
# → fields.String(required=False)
# Column with length constraint gets Length validator
username = sa.Column(sa.String(50))
# → fields.String(validate=[validate.Length(max=50)])
# Enum column gets OneOf validator
priority = sa.Column(sa.Enum("low", "medium", "high"))
# → fields.String(validate=[validate.OneOf(["low", "medium", "high"])])
# Foreign key columns are excluded by default (unless include_fk=True)
user_id = sa.Column(sa.Integer, sa.ForeignKey("users.id"))
# → Excluded unless include_fk=True
# Relationship properties become Related fields
posts = relationship("Post", back_populates="author")
# → Related() field (unless include_relationships=False)Install with Tessl CLI
npx tessl i tessl/pypi-marshmallow-sqlalchemy