QuerySet¶

async def execute(self) -> typing.Any:
    raise NotImplementedError()

def _update_auto_now_fields(self, values: Any, fields: Any) -> Any:
    """Refresh `auto_now` date and datetime fields inside a payload.

    Returns:
        Any: Updated payload dictionary.
    """
    for k, v in fields.items():
        if isinstance(v, (DateField, DateTimeField)) and v.auto_now:
            values[k] = v.validator.get_default_value()  # type: ignore
    return values

def _resolve_value(self, value: typing.Any) -> typing.Any:
    """Normalize one value for storage or query generation.

    Dictionaries are JSON-encoded, enums are converted to their member
    names, and every other value is returned unchanged.

    Args:
        value (typing.Any): Raw value supplied by model or queryset code.

    Returns:
        typing.Any: Normalized value suitable for downstream processing.
    """
    if isinstance(value, dict):
        return dumps(
            value,
            option=OPT_SERIALIZE_NUMPY | OPT_OMIT_MICROSECONDS,
        ).decode("utf-8")
    elif isinstance(value, Enum):
        return value.name
    return value

def prefetch_related(self, *prefetch: Prefetch) -> "QuerySet":
    """
    Performs a reverse lookup for the foreignkeys. This is different
    from the select_related. Select related performs a follows the SQL foreign relation
    whereas the prefetch_related follows the python relations.
    """
    queryset: QuerySet = self._clone()

    if not isinstance(prefetch, (list, tuple)):
        prefetch = cast("list[Prefetch]", [prefetch])  # type: ignore

    if isinstance(prefetch, tuple):
        prefetch = list(prefetch)  # type: ignore

    if any(not isinstance(value, Prefetch) for value in prefetch):
        raise QuerySetError("The prefetch_related must have Prefetch type objects only.")

    prefetch = list(self._prefetch_related) + prefetch  # type: ignore
    queryset._prefetch_related = prefetch
    return queryset

def using(
    self,
    _positional: Any = _sentinel,
    *,
    database: str | Database | None | Any = _undefined,
    schema: str | None | bool | Any = _undefined,
) -> QuerySet:
    if _positional is not _sentinel:
        warnings.warn(
            "Passing positional arguments to using is deprecated. Use schema= instead.",
            DeprecationWarning,
            stacklevel=2,
        )
        schema = _positional

    queryset = cast("QuerySet", self._clone())

    if database is not _undefined:
        if isinstance(database, Database):
            queryset.database = database
        elif database is None:
            queryset.database = self.model_class.meta.registry.database
        else:
            assert database in self.model_class.meta.registry.extra, (
                f"`{database}` is not in the connections extra of the model"
                f"`{self.model_class.__name__}` registry"
            )
            queryset.database = self.model_class.meta.registry.extra[database]

    if schema is not _undefined:
        queryset.using_schema = None if schema is False else schema
        if schema is False:
            queryset.table = self.model_class.table
        else:
            queryset.table = self.model_class.table_schema(
                cast("str | None", queryset.using_schema)
            )

    return queryset

def using_with_db(
    self, connection_name: str, schema: str | None | bool | Any = _undefined
) -> QuerySet:
    warnings.warn(
        "'using_with_db' is deprecated in favor of 'using' with schema, database arguments.",
        DeprecationWarning,
        stacklevel=2,
    )
    return self.using(database=connection_name, schema=schema)

def _clear_cache(
    self,
    *,
    keep_result_cache: bool = False,
    keep_cached_selected: bool = False,
) -> None:
    if not keep_result_cache:
        self._cache.clear()
    if not keep_cached_selected:
        self._cached_select_with_tables = None
    self._cache_count: int | None = None
    self._cache_first: SaffierModel | None = None
    self._cache_last: SaffierModel | None = None
    self._cache_fetch_all = False

def _cache_or_return_result(self, result: SaffierModel) -> SaffierModel:
    cached = self._cache.get(self.model_class, result)
    if cached is not None:
        return cast("SaffierModel", cached)
    self._cache.update(self.model_class, [result])
    return result

def _build_order_by_expression(
    self,
    order_by: Any,
    expression: Any,
    tables_and_models: dict[str, tuple[Any, Any]],
) -> Any:
    """Apply ordering expressions to a SQLAlchemy selectable.

    Args:
        order_by: Raw order-by field paths.
        expression: Selectable being modified.
        tables_and_models: Join map used to resolve relationship paths.

    Returns:
        Any: Updated selectable.
    """
    order_by = [self._prepare_order_by(value, tables_and_models) for value in order_by]
    expression = expression.order_by(*order_by)
    return expression

def _build_group_by_expression(
    self,
    group_by: Any,
    expression: Any,
    tables_and_models: dict[str, tuple[Any, Any]],
) -> Any:
    """Apply group-by expressions to a SQLAlchemy selectable.

    Args:
        group_by: Raw group-by field paths.
        expression: Selectable being modified.
        tables_and_models: Join map used to resolve relationship paths.

    Returns:
        Any: Updated selectable.
    """
    group_by = [self._prepare_group_by(value, tables_and_models) for value in group_by]
    expression = expression.group_by(*group_by)
    return expression

def _build_filter_clauses_expression(self, filter_clauses: Any, expression: Any) -> Any:
    """Apply AND-combined filter clauses to a selectable.

    Returns:
        Any: Updated selectable.
    """
    if len(filter_clauses) == 1:
        clause = filter_clauses[0]
    else:
        clause = sqlalchemy.sql.and_(*filter_clauses)
    expression = expression.where(clause)
    return expression

def _build_or_clauses_expression(self, or_clauses: Any, expression: Any) -> Any:
    """Apply OR-combined clauses to a selectable.

    Returns:
        Any: Updated selectable.
    """
    clause = or_clauses[0] if len(or_clauses) == 1 else sqlalchemy.sql.or_(*or_clauses)
    expression = expression.where(clause)
    return expression

def _build_select_distinct(
    self,
    distinct_on: Any,
    expression: Any,
    tables_and_models: dict[str, tuple[Any, Any]],
) -> Any:
    """Apply `DISTINCT` or `DISTINCT ON` semantics to a selectable.

    Returns:
        Any: Updated selectable.
    """
    if not distinct_on:
        expression = expression.distinct()
    else:
        distinct_on = [
            self._prepare_fields_for_distinct(value, tables_and_models)
            for value in distinct_on
        ]
        expression = expression.distinct(*distinct_on)
    return expression

def _is_multiple_foreign_key(
    self, model_class: type["Model"] | type["ReflectModel"]
) -> tuple[bool, list[tuple[str, str, str]]]:
    """
    Checks if the table has multiple foreign keys to the same table.
    Iterates through all fields of type ForeignKey and OneToOneField and checks if there are
    multiple FKs to the same destination table.

    Returns a tuple of bool, list of tuples
    """
    fields = model_class.fields  # type: ignore
    foreign_keys = []
    has_many = False

    counter = 0

    for key, value in fields.items():
        tablename = None

        if counter > 1:
            has_many = True

        if isinstance(value, (saffier.ForeignKey, saffier.OneToOneField)):
            tablename = value.to if isinstance(value.to, str) else value.to.__name__

            if tablename not in foreign_keys:
                foreign_keys.append((key, tablename, value.related_name))
                counter += 1
            else:
                counter += 1

    return has_many, foreign_keys  # type: ignore

def _resolve_many_to_many_join_fields(
    self,
    previous_model_class: type["Model"] | type["ReflectModel"],
    related_field: saffier_fields.ManyToManyField,
) -> tuple[type["Model"] | type["ReflectModel"], str, str]:
    through_model = related_field.through
    if isinstance(through_model, str):
        registry = getattr(previous_model_class.meta, "registry", None)
        if registry is not None:
            through_model = registry.models.get(through_model) or registry.reflected.get(
                through_model
            )
            related_field.through = through_model

    if through_model is None or isinstance(through_model, str):
        raise QuerySetError(
            detail=(
                "Unable to resolve through model for select_related path "
                f"'{previous_model_class.__name__}.{related_field.name}'."
            )
        )

    from_field: str | None = None
    to_field: str | None = None
    for field_name, field in through_model.fields.items():
        if not isinstance(field, (saffier_fields.ForeignKey, saffier_fields.OneToOneField)):
            continue
        if field.target is previous_model_class and from_field is None:
            from_field = field_name
            continue
        if field.target is related_field.target and to_field is None:
            to_field = field_name

    if from_field is None or to_field is None:
        raise QuerySetError(
            detail=(
                "Unable to resolve many-to-many join fields for "
                f"'{previous_model_class.__name__}.{related_field.name}'."
            )
        )

    return cast("type[Model] | type[ReflectModel]", through_model), from_field, to_field

@staticmethod
def _primary_join_columns(
    table: Any, model_class: type["Model"] | type["ReflectModel"]
) -> list[Any]:
    columns = [
        getattr(table.c, pk_name, None) for pk_name in getattr(model_class, "pkcolumns", ())
    ]
    resolved = [column for column in columns if column is not None]
    if resolved:
        return resolved
    return list(table.primary_key)

@staticmethod
def _relation_join_columns(table: Any, field_name: str, field: Any) -> list[Any]:
    column_names = (
        field.get_column_names(field_name)
        if hasattr(field, "get_column_names")
        else (field_name,)
    )
    return [getattr(table.c, column_name, None) for column_name in column_names]

@staticmethod
def _target_relation_columns(table: Any, field: Any) -> list[Any]:
    related_columns = getattr(field, "related_columns", {})
    return [getattr(table.c, column_name, None) for column_name in related_columns]

@staticmethod
def _build_join_condition(left_columns: Sequence[Any], right_columns: Sequence[Any]) -> Any:
    pairs = [
        (left_column, right_column)
        for left_column, right_column in zip(left_columns, right_columns, strict=False)
        if left_column is not None and right_column is not None
    ]
    if not pairs or len(pairs) != len(left_columns) or len(pairs) != len(right_columns):
        raise QuerySetError(detail="Unable to resolve join columns for relationship.")
    clauses = [left_column == right_column for left_column, right_column in pairs]
    if len(clauses) == 1:
        return clauses[0]
    return sqlalchemy.and_(*clauses)

def _dedupe_related_paths(self, paths: Sequence[str]) -> list[str]:
    related_paths = list(dict.fromkeys(path for path in paths if path))
    return [
        path
        for path in related_paths
        if not any(other != path and other.startswith(f"{path}__") for other in related_paths)
    ]

def _collect_related_paths(self, fields: Sequence[str]) -> list[str]:
    from saffier.core.db.relationships.utils import crawl_relationship

    related_paths: list[str] = []
    for value in fields:
        if not isinstance(value, str):
            continue
        crawl_result = crawl_relationship(
            self.model_class,
            value.lstrip("-"),
            model_database=self.database,
        )
        if crawl_result.forward_path:
            related_paths.append(crawl_result.forward_path)
    return self._dedupe_related_paths(related_paths)

def _build_tables_select_from_relationship(
    self,
    related_paths: Sequence[str] | None = None,
) -> Any:
    """
    Builds the tables relationships and joins.
    When a table contains more than one foreign key pointing to the same
    destination table, a lookup for the related field is made to understand
    from which foreign key the table is looked up from.
    """
    queryset: QuerySet = self._clone()

    tables = [queryset.table]
    select_from = queryset.table
    tables_and_models: dict[str, tuple[Any, Any]] = {
        "": (queryset.table, queryset.model_class)
    }

    select_related = self._dedupe_related_paths(
        queryset._select_related if related_paths is None else related_paths
    )

    # Select related
    for item in select_related:
        # For m2m relationships
        model_class = queryset.model_class
        current_table = queryset.table
        prefix = ""

        for part in item.split("__"):
            has_many_fk_same_table = False
            keys: list[tuple[str, str, str]] = []
            previous_model_class = model_class
            previous_table = current_table
            join_lookup_field: str | None = None
            reverse_join = False
            current_related_field: Any = None
            through_model: Any = None
            through_table = None
            through_from_field: str | None = None
            through_to_field: str | None = None
            try:
                current_related_field = model_class.fields[part]
                model_class = current_related_field.target
                if isinstance(
                    current_related_field,
                    (saffier_fields.ForeignKey, saffier_fields.OneToOneField),
                ):
                    join_lookup_field = part
                elif isinstance(current_related_field, saffier_fields.ManyToManyField):
                    through_model, through_from_field, through_to_field = (
                        self._resolve_many_to_many_join_fields(
                            previous_model_class,
                            current_related_field,
                        )
                    )
                    through_table = (
                        through_model.table_schema(queryset.using_schema)
                        if queryset.using_schema is not None
                        else through_model.table
                    )
            except (KeyError, AttributeError):
                # Check related fields
                model_class = getattr(model_class, part).related_from
                reverse_join = True
                join_lookup_field = previous_model_class.meta.related_names_mapping.get(part)
                has_many_fk_same_table, keys = self._is_multiple_foreign_key(model_class)

            next_prefix = part if not prefix else f"{prefix}__{part}"

            if queryset.using_schema is not None:
                table = model_class.table_schema(queryset.using_schema)
            else:
                table = model_class.table
            if model_class is previous_model_class or next_prefix in tables_and_models:
                table = table.alias(
                    hash_tablekey(tablekey=model_class.meta.tablename, prefix=next_prefix)
                )

            if isinstance(current_related_field, saffier_fields.ManyToManyField):
                assert through_table is not None
                assert through_model is not None
                through_from_relation = through_model.fields[through_from_field]
                through_to_relation = through_model.fields[through_to_field]
                through_prefix = f"{next_prefix}__through"
                if (
                    through_model is previous_model_class
                    or through_prefix in tables_and_models
                ):
                    through_table = through_table.alias(
                        hash_tablekey(
                            tablekey=through_model.meta.tablename,
                            prefix=through_prefix,
                        )
                    )
                left_columns = self._primary_join_columns(previous_table, previous_model_class)
                through_left_columns = self._relation_join_columns(
                    through_table,
                    through_from_field,
                    through_from_relation,
                )
                through_right_columns = self._relation_join_columns(
                    through_table,
                    through_to_field,
                    through_to_relation,
                )
                right_columns = self._primary_join_columns(table, model_class)
                select_from = sqlalchemy.sql.join(
                    select_from,
                    through_table,
                    self._build_join_condition(left_columns, through_left_columns),
                )
                select_from = sqlalchemy.sql.join(
                    select_from,
                    table,
                    self._build_join_condition(through_right_columns, right_columns),
                )
                prefix = next_prefix
                tables_and_models[prefix] = (table, model_class)
                current_table = table
                tables.append(table)
                continue

            # If there is multiple FKs to the same table
            if not has_many_fk_same_table:
                if join_lookup_field is not None:
                    if reverse_join:
                        relation_field = model_class.fields[join_lookup_field]
                        left_columns = self._target_relation_columns(
                            previous_table,
                            relation_field,
                        )
                        right_columns = self._relation_join_columns(
                            table,
                            join_lookup_field,
                            relation_field,
                        )
                    else:
                        relation_field = current_related_field
                        left_columns = self._relation_join_columns(
                            previous_table,
                            join_lookup_field,
                            relation_field,
                        )
                        right_columns = self._target_relation_columns(table, relation_field)

                    select_from = sqlalchemy.sql.join(
                        select_from,
                        table,
                        self._build_join_condition(left_columns, right_columns),
                    )
                else:
                    select_from = sqlalchemy.sql.join(select_from, table)
            else:
                lookup_field = None

                # Extract the table field from the related
                # Assign to a lookup_field
                for values in keys:
                    field, _, related_name = values
                    if related_name == part:
                        lookup_field = field
                        break

                if lookup_field is None:
                    select_from = sqlalchemy.sql.join(select_from, table)
                else:
                    relation_field = model_class.fields[lookup_field]
                    left_columns = self._target_relation_columns(
                        previous_table, relation_field
                    )
                    right_columns = self._relation_join_columns(
                        table,
                        lookup_field,
                        relation_field,
                    )
                    select_from = sqlalchemy.sql.join(
                        select_from,
                        table,
                        self._build_join_condition(left_columns, right_columns),
                    )

            prefix = next_prefix
            tables_and_models[prefix] = (table, model_class)
            current_table = table
            tables.append(table)
    return tables, select_from, tables_and_models

def _should_include_selected_column(
    self,
    field_name: str,
    model_class: Any,
    prefix: str = "",
) -> bool:
    if self._only:
        if not prefix and field_name not in self._only:
            return False
        if prefix and prefix not in self._only and f"{prefix}__{field_name}" not in self._only:
            return False

    if self._defer:
        if not prefix and field_name in self._defer:
            return False
        if prefix and (prefix in self._defer or f"{prefix}__{field_name}" in self._defer):
            return False

    return not (
        self._exclude_secrets
        and field_name in model_class.meta.fields
        and model_class.meta.fields[field_name].secret
    )

def _build_select_columns(
    self,
    tables_and_models: dict[str, tuple[Any, Any]],
    selectable_related: set[str],
) -> list[Any]:
    columns: list[Any] = [*self._extra_select]

    for prefix, (table, model_class) in tables_and_models.items():
        if prefix and not any(
            related_path == prefix or related_path.startswith(f"{prefix}__")
            for related_path in selectable_related
        ):
            continue

        for column_key, column in table.columns.items():
            field_name = model_class.meta.columns_to_field.get(column_key, column_key)
            if not self._should_include_selected_column(field_name, model_class, prefix):
                continue

            columns.append(column)

    if not columns:
        raise QuerySetError("No columns selected for queryset.")
    return columns

def _build_where_clause(
    self,
    outer_tables_and_models: dict[str, tuple[Any, Any]],
    outer_select_paths: Sequence[str],
) -> Any:
    where_clauses: list[Any] = []
    if self.or_clauses:
        clause = (
            self.or_clauses[0]
            if len(self.or_clauses) == 1
            else sqlalchemy.sql.or_(*self.or_clauses)
        )
        where_clauses.append(clause)
    if self.filter_clauses:
        where_clauses.extend(self.filter_clauses)
    if not where_clauses:
        return None

    full_select_paths = self._dedupe_related_paths(
        [*outer_select_paths, *self._filter_related]
    )
    _, full_select_from, full_tables_and_models = self._build_tables_select_from_relationship(
        full_select_paths
    )

    if len(full_tables_and_models) == 1:
        return (
            where_clauses[0]
            if len(where_clauses) == 1
            else sqlalchemy.sql.and_(*where_clauses)
        )

    outer_table, outer_model = outer_tables_and_models[""]
    pk_columns = [getattr(outer_table.c, column) for column in outer_model.pkcolumns]
    subquery = (
        sqlalchemy.select(*pk_columns).select_from(full_select_from).where(*where_clauses)
    )

    if len(pk_columns) == 1:
        return pk_columns[0].in_(subquery)
    return sqlalchemy.tuple_(*pk_columns).in_(subquery)

def _validate_only_and_defer(self) -> None:
    if self._only and self._defer:
        raise QuerySetError("You cannot use .only() and .defer() at the same time.")

def _secret_recursive_names(
    self, model_class: Any, columns: list[str] | None = None
) -> list[str]:
    """Collect non-secret field names across a model graph.

    Foreign-key targets are traversed recursively so secret filtering can be
    applied consistently to nested serialization shapes.

    Args:
        model_class: Model whose fields should be inspected.
        columns: Internal accumulator used during recursion.

    Returns:
        list[str]: Distinct field names that are not marked as secret.
    """
    if columns is None:
        columns = []

    for name, field in model_class.fields.items():
        if isinstance(field, saffier_fields.ForeignKey):
            # Making sure the foreign key is always added unless is a secret
            if not field.secret:
                columns.append(name)
                columns.extend(
                    self._secret_recursive_names(model_class=field.target, columns=columns)
                )
            continue
        if not field.secret:
            columns.append(name)

    columns = list(set(columns))
    return columns

def _build_select_with_tables(self) -> tuple[Any, dict[str, tuple[Any, Any]]]:
    """Build the main select statement together with its join map.

    The method resolves every relationship path needed by filtering,
    ordering, grouping, or explicit `select_related()`, applies `WHERE`,
    `ORDER BY`, `GROUP BY`, `DISTINCT`, pagination, and row-locking options,
    then caches the resulting expression for reuse.

    Returns:
        tuple[Any, dict[str, tuple[Any, Any]]]: SQLAlchemy select expression
        plus the table/model map used to hydrate rows.
    """
    if self._cached_select_with_tables is not None:
        return self._cached_select_with_tables

    queryset = self

    queryset._validate_only_and_defer()
    outer_select_paths = queryset._dedupe_related_paths(
        [
            *queryset._select_related,
            *queryset._collect_related_paths(queryset._order_by),
            *queryset._collect_related_paths(queryset._group_by),
        ]
    )
    _, select_from, tables_and_models = queryset._build_tables_select_from_relationship(
        outer_select_paths
    )
    selectable_related = set(queryset._select_related)
    columns = queryset._build_select_columns(tables_and_models, selectable_related)
    expression = sqlalchemy.sql.select(*columns).select_from(select_from)

    where_clause = queryset._build_where_clause(tables_and_models, outer_select_paths)
    if where_clause is not None:
        expression = expression.where(where_clause)

    if queryset._order_by:
        expression = queryset._build_order_by_expression(
            queryset._order_by,
            expression=expression,
            tables_and_models=tables_and_models,
        )

    if queryset.limit_count:
        expression = expression.limit(queryset.limit_count)

    if queryset._offset:
        expression = expression.offset(queryset._offset)

    if queryset._group_by:
        expression = queryset._build_group_by_expression(
            queryset._group_by,
            expression=expression,
            tables_and_models=tables_and_models,
        )

    if queryset.distinct_on is not None:
        expression = queryset._build_select_distinct(
            queryset.distinct_on,
            expression=expression,
            tables_and_models=tables_and_models,
        )

    if queryset._for_update:
        for_update = dict(queryset._for_update)
        of = for_update.get("of")
        if of:
            for_update["of"] = tuple(
                getattr(model, "table", model) for model in cast("tuple[Any, ...]", of)
            )
        expression = expression.with_for_update(**for_update)

    queryset._expression = expression  # type: ignore
    if queryset._select_related:
        queryset._cached_select_related_expression = expression
        queryset._source_queryset._cached_select_related_expression = expression
    queryset._cached_select_with_tables = (expression, tables_and_models)
    return queryset._cached_select_with_tables

def _build_select(self) -> Any:
    return self._build_select_with_tables()[0]

async def _hydrate_row(
    self,
    queryset: "QuerySet",
    row: Any,
    tables_and_models: dict[str, tuple[Any, Any]],
    *,
    is_only_fields: bool,
    is_defer_fields: bool,
) -> SaffierModel:
    result = queryset.model_class.from_query_result(
        row,
        select_related=queryset._select_related,
        prefetch_related=[],
        is_only_fields=is_only_fields,
        only_fields=queryset._only,
        is_defer_fields=is_defer_fields,
        using_schema=queryset.using_schema,
        exclude_secrets=queryset._exclude_secrets,
        reference_select=queryset._reference_select,
        tables_and_models=tables_and_models,
    )
    if queryset._prefetch_related:
        result = await queryset.model_class.apply_prefetch_related(
            row=row,
            model=result,
            prefetch_related=queryset._prefetch_related,
        )
    return result

def _filter_query(self, exclude: bool = False, or_: bool = False, **kwargs: Any) -> "QuerySet":
    clauses: list[Any] = []
    filter_clauses = self.filter_clauses
    or_clauses = self.or_clauses
    select_related = list(self._select_related)
    filter_related = list(self._filter_related)
    prefetch_related = list(self._prefetch_related)

    # Making sure for queries we use the main class and not the proxy
    # And enable the parent
    if self.model_class.is_proxy_model:
        self.model_class = self.model_class.parent

    clauses, implied_select_related = build_lookup_clauses(
        self.model_class,
        self.table,
        kwargs,
        escape_characters=tuple(self.ESCAPE_CHARACTERS),
        using_schema=self.using_schema,
        embed_parent=self.embed_parent_filters,
        model_database=self.database,
    )
    for related_path in implied_select_related:
        if related_path not in filter_related:
            filter_related.append(related_path)

    if exclude:
        if clauses:
            if not or_:
                filter_clauses.append(sqlalchemy.not_(sqlalchemy.sql.and_(*clauses)))
            else:
                or_clauses.append(sqlalchemy.not_(sqlalchemy.sql.and_(*clauses)))
    else:
        if not or_:
            filter_clauses += clauses
        else:
            or_clauses += clauses

    return cast(
        "QuerySet",
        self.__class__(
            model_class=self.model_class,
            database=self._database,
            filter_clauses=filter_clauses,
            or_clauses=or_clauses,
            select_related=select_related,
            filter_related=filter_related,
            prefetch_related=prefetch_related,
            limit_count=self.limit_count,
            limit_offset=self._offset,
            order_by=self._order_by,
            only_fields=self._only,
            defer_fields=self._defer,
            m2m_related=self.m2m_related,
            exclude_secrets=self._exclude_secrets,
            table=self.table,
            using_schema=self.using_schema,
            for_update=self._for_update,
            batch_size=self._batch_size,
            extra_select=self._extra_select,
            reference_select=self._reference_select,
            embed_parent=self.embed_parent,
        ),
    )

def _validate_kwargs(self, **kwargs: Any) -> Any:
    original_kwargs = dict(kwargs)
    fields = self.model_class.fields
    for key, field in fields.items():
        field.modify_input(key, kwargs)
    schema_fields = {}
    for key, value in fields.items():
        if not value.has_column():
            continue
        field_validator = value.validator
        if key in kwargs and field_validator.read_only:
            field_validator = copy.copy(field_validator)
            field_validator.read_only = False
        schema_fields[key] = field_validator
    validator = Schema(fields=schema_fields)
    kwargs = validator.check(kwargs)
    for key, value in original_kwargs.items():
        field = fields.get(key)
        if field is not None and not field.has_column():
            kwargs[key] = value
    for key, value in fields.items():
        if not value.has_column():
            continue
        if value.validator.read_only and value.validator.has_default():
            kwargs[key] = value.validator.get_default_value()
    return kwargs

def _normalize_many_to_many_values(
    self,
    field: saffier_fields.ManyToManyField,
    value: Any,
) -> list[Any]:
    if value is None:
        return []

    values: Sequence[Any]
    if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
        values = value
    else:
        values = [value]

    target = field.target
    normalized = []
    for item in values:
        if item is None:
            continue
        if isinstance(item, target):
            normalized.append(item)
        else:
            normalized.append(target(pk=item))
    return normalized