feat(service): add configurable schema dump settings (#750)

## Problem

Service model conversion currently hardcodes partial-update dump
behavior for schema-like inputs. Pydantic models are dumped with
`exclude_unset=True`, msgspec structs drop `UNSET`, and attrs instances
drop `attrs.NOTHING`. That preserves partial-update semantics, but it
gives callers no supported way to intentionally include unset/default
values for update-like schemas.

A common case is a schema where an unset field should still be applied
with its schema default:

```python
class InviteUpdate(BaseModel):
    name: str | None = None
    is_admin: bool = False

await invite_service.update(data, item_id=invite_id)
```

With the previous hardcoded behavior, `is_admin=False` is excluded
unless the caller explicitly sets it on the Pydantic model.

## Changes

### `feat(service): add configurable schema dump behavior`

Adds `SchemaDumpConfig` and exposes it as a class-level service default
plus a per-call override on write operations that convert schema-like
input through `to_model()`.

```python
class InviteService(SQLAlchemyAsyncRepositoryService[Invite]):
    schema_dump_config = SchemaDumpConfig(exclude_unset=False)

await invite_service.update(
    data,
    item_id=invite_id,
    schema_dump_config=SchemaDumpConfig(exclude_unset=False),
)
```

The per-call override is scoped to that operation only and does not
mutate the service default or affect later calls.

### `feat(serialization): centralize schema dumping`

Routes Pydantic, msgspec, attrs, and dataclass inputs through
`schema_dump(..., config=...)` instead of duplicating hardcoded
conversion logic inside sync and async services.

`SchemaDumpConfig` supports:

- `exclude_unset`, for Pydantic field-set behavior
- `exclude_none`, for dropping `None` values
- `exclude_defaults`, for dropping values equal to schema defaults where
supported
- `exclude_sentinels`, for dropping sentinel values such as Pydantic
`MISSING`, msgspec `UNSET`, attrs `NOTHING`, and Advanced Alchemy
`Empty`

### `fix(fixtures): align service overrides with the new conversion
contract`

Updates test fixture services that override `to_model()` so they accept
and forward `schema_dump_config`. This keeps custom service conversion
hooks compatible with the new base method signature and preserves the
selected dump behavior in the inline slug-generation path.

## Compatibility Notes

- Existing defaults preserve current partial-update behavior:
`SchemaDumpConfig()` keeps `exclude_unset=True` and
`exclude_sentinels=True`.
- The existing `schema_dump(..., exclude_unset=...)` argument remains
supported for callers using the standalone utility.
- Optional schema libraries remain guarded. Advanced Alchemy should not
fail to import when Pydantic, msgspec, attrs, or cattrs are absent.
- `to_schema()` is not changed by this PR; the configurable behavior
applies to service input conversion through `to_model()`.

Author: cofin

advanced_alchemy/service/__init__.py

@@ -28,6 +28,7 @@
     ModelDictListT,
     ModelDictT,
     ModelDTOT,
+    SchemaDumpConfig,
     SupportedSchemaModel,
     fields,
     is_attrs_instance,
@@ -77,6 +78,7 @@
     "SQLAlchemySyncQueryService",
     "SQLAlchemySyncRepositoryReadService",
     "SQLAlchemySyncRepositoryService",
+    "SchemaDumpConfig",
     "SupportedSchemaModel",
     "fields",
     "find_filter",

advanced_alchemy/service/_async.py

@@ -31,18 +31,18 @@
 from advanced_alchemy.utils.dataclass import Empty, EmptyType
 from advanced_alchemy.utils.deprecation import warn_deprecation
 from advanced_alchemy.utils.serialization import (
-    UNSET,
     BulkModelDictT,
     ModelDictListT,
     ModelDictT,
-    asdict,
-    attrs_nothing,
+    SchemaDumpConfig,
     is_attrs_instance,
+    is_dataclass,
     is_dict,
     is_dto_data,
     is_msgspec_struct,
     is_pydantic_model,
     is_sqlmodel_table_model,
+    schema_dump,
 )
 
 
@@ -107,6 +107,8 @@ class SQLAlchemyAsyncRepositoryReadService(ResultConverter, Generic[ModelT, SQLA
     """Optionally apply the ``unique()`` method to results before returning."""
     count_with_window_function: ClassVar[bool] = True
     """Use an analytical window function to count results.  This allows the count to be performed in a single query."""
+    schema_dump_config: ClassVar[SchemaDumpConfig] = SchemaDumpConfig()
+    """Default dump behavior when schema-like input is converted into a model."""
     _repository_instance: SQLAlchemyAsyncRepositoryT
 
     def __init__(
@@ -462,16 +464,19 @@ async def to_model(
         self,
         data: "ModelDictT[ModelT]",
         operation: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> ModelT:
         """Parse and Convert input into a model.
 
         Args:
             data: Representations to be created.
             operation: Optional operation flag so that you can provide behavior based on CRUD operation
+            schema_dump_config: Optional schema dump behavior for this conversion.
 
         Returns:
             Representation of created instances.
         """
+        dump_config = schema_dump_config if schema_dump_config is not None else self.schema_dump_config
         operation_map = {
             "create": self.to_model_on_create,
             "update": self.to_model_on_update,
@@ -486,35 +491,15 @@ async def to_model(
             return model_from_dict(self.model_type, **data)
         if is_sqlmodel_table_model(data):
             return model_from_dict(self.model_type, **model_to_dict(cast("ModelProtocol", data)))
-        if is_pydantic_model(data):
+        if is_pydantic_model(data) or is_msgspec_struct(data) or is_attrs_instance(data) or is_dataclass(data):
             return model_from_dict(
                 self.model_type,
-                **data.model_dump(exclude_unset=True),
-            )
-
-        if is_msgspec_struct(data):
-            return model_from_dict(
-                self.model_type,
-                **{
-                    f: getattr(data, f)
-                    for f in data.__struct_fields__
-                    if hasattr(data, f) and getattr(data, f) is not UNSET
-                },
+                **schema_dump(data, config=dump_config),
             )
 
         if is_dto_data(data):
             return cast("ModelT", data.create_instance())
 
-        if is_attrs_instance(data):
-            # Filter out attrs.NOTHING values for partial updates
-            def filter_unset(attr: Any, value: Any) -> bool:  # noqa: ARG001
-                return value is not attrs_nothing
-
-            return model_from_dict(
-                self.model_type,
-                **asdict(data, filter=filter_unset),
-            )
-
         # Fallback for objects with __dict__ (e.g., regular classes)
         if hasattr(data, "__dict__") and not isinstance(data, self.model_type):
             return model_from_dict(
@@ -773,6 +758,7 @@ async def create(
         auto_refresh: Optional[bool] = None,
         error_messages: Optional[Union[ErrorMessages, EmptyType]] = Empty,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> "ModelT":
         """Wrap repository instance creation.
 
@@ -784,11 +770,12 @@ async def create(
             error_messages: An optional dictionary of templates to use
                 for friendlier error messages to clients
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Returns:
             Representation of created instance.
         """
-        data = await self.to_model(data, "create")
+        data = await self.to_model(data, "create", schema_dump_config=schema_dump_config)
         return cast(
             "ModelT",
             await self.repository.add(
@@ -809,6 +796,7 @@ async def create_many(
         auto_expunge: Optional[bool] = None,
         error_messages: Optional[Union[ErrorMessages, EmptyType]] = Empty,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> Sequence[ModelT]:
         """Wrap repository bulk instance creation.
 
@@ -819,13 +807,17 @@ async def create_many(
             error_messages: An optional dictionary of templates to use
                 for friendlier error messages to clients
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Returns:
             Representation of created instances.
         """
         if is_dto_data(data):
             data = data.create_instance()
-        data = [(await self.to_model(datum, "create")) for datum in cast("ModelDictListT[ModelT]", data)]
+        data = [
+            (await self.to_model(datum, "create", schema_dump_config=schema_dump_config))
+            for datum in cast("ModelDictListT[ModelT]", data)
+        ]
         return cast(
             "Sequence[ModelT]",
             await self.repository.add_many(
@@ -853,6 +845,7 @@ async def update(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> "ModelT":
         """Wrap repository update operation.
 
@@ -875,6 +868,7 @@ async def update(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Raises:
             RepositoryError: If no configuration or session is provided.
@@ -885,7 +879,7 @@ async def update(
         # ALWAYS convert data through to_model first to ensure operation hooks are called
         # This ensures custom to_model() implementations receive the operation="update" parameter
         # and that to_model_on_update() is properly invoked via the operation_map
-        data = await self.to_model(data, "update")
+        data = await self.to_model(data, "update", schema_dump_config=schema_dump_config)
 
         if item_id is not None:
             # When item_id is provided, update existing instance rather than replacing it
@@ -958,6 +952,7 @@ async def update_many(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> Sequence[ModelT]:
         """Wrap repository bulk instance update.
 
@@ -971,13 +966,17 @@ async def update_many(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Returns:
             Representation of updated instances.
         """
         if is_dto_data(data):
             data = data.create_instance()
-        data = [(await self.to_model(datum, "update")) for datum in cast("ModelDictListT[ModelT]", data)]
+        data = [
+            (await self.to_model(datum, "update", schema_dump_config=schema_dump_config))
+            for datum in cast("ModelDictListT[ModelT]", data)
+        ]
         return cast(
             "Sequence[ModelT]",
             await self.repository.update_many(
@@ -1008,6 +1007,7 @@ async def upsert(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> ModelT:
         """Wrap repository upsert operation.
 
@@ -1031,11 +1031,12 @@ async def upsert(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Returns:
             Updated or created representation.
         """
-        data = await self.to_model(data, "upsert")
+        data = await self.to_model(data, "upsert", schema_dump_config=schema_dump_config)
         # Handle ID extraction and setting for single-column PKs
         # For composite PKs, the repository's upsert() handles this directly
         if item_id is None:
@@ -1079,6 +1080,7 @@ async def upsert_many(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
     ) -> Sequence[ModelT]:
         """Wrap repository upsert operation.
 
@@ -1095,13 +1097,17 @@ async def upsert_many(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
 
         Returns:
             Updated or created representation.
         """
         if is_dto_data(data):
             data = data.create_instance()
-        data = [(await self.to_model(datum, "upsert")) for datum in cast("ModelDictListT[ModelT]", data)]
+        data = [
+            (await self.to_model(datum, "upsert", schema_dump_config=schema_dump_config))
+            for datum in cast("ModelDictListT[ModelT]", data)
+        ]
         return cast(
             "Sequence[ModelT]",
             await self.repository.upsert_many(
@@ -1133,6 +1139,7 @@ async def get_or_upsert(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
         **kwargs: Any,
     ) -> tuple[ModelT, bool]:
         """Wrap repository instance creation.
@@ -1158,13 +1165,14 @@ async def get_or_upsert(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
             **kwargs: Identifier of the instance to be retrieved.
 
         Returns:
             Representation of created instance.
         """
         match_fields = match_fields or self.match_fields
-        validated_model = await self.to_model(kwargs, "create")
+        validated_model = await self.to_model(kwargs, "create", schema_dump_config=schema_dump_config)
         return cast(
             "tuple[ModelT, bool]",
             await self.repository.get_or_upsert(
@@ -1199,6 +1207,7 @@ async def get_and_update(
         execution_options: Optional[dict[str, Any]] = None,
         uniquify: Optional[bool] = None,
         bind_group: Optional[str] = None,
+        schema_dump_config: Optional[SchemaDumpConfig] = None,
         **kwargs: Any,
     ) -> tuple[ModelT, bool]:
         """Wrap repository instance creation.
@@ -1221,13 +1230,14 @@ async def get_and_update(
             execution_options: Set default execution options
             uniquify: Optionally apply the ``unique()`` method to results before returning.
             bind_group: Optional routing group to use for the operation.
+            schema_dump_config: Optional schema dump behavior for this operation.
             **kwargs: Identifier of the instance to be retrieved.
 
         Returns:
             Representation of updated instance.
         """
         match_fields = match_fields or self.match_fields
-        validated_model = await self.to_model(kwargs, "update")
+        validated_model = await self.to_model(kwargs, "update", schema_dump_config=schema_dump_config)
         return cast(
             "tuple[ModelT, bool]",
             await self.repository.get_and_update(