OneToOneField¶

def get_validator(self, **kwargs: typing.Any) -> SaffierField:
    return self.ForeignKeyValidator(**kwargs)

def get_column_type(self) -> sqlalchemy.types.TypeEngine:
    raise NotImplementedError()  # pragma: no cover

def get_constraints(self) -> typing.Any:
    return []

def get_columns(self, name: str) -> typing.Sequence[sqlalchemy.Column]:
    target = self.target
    columns: list[sqlalchemy.Column] = []
    for related_key, related_column in self.related_columns.items():
        related_field = target.fields[related_key]
        related_type = (
            related_column.type
            if related_column is not None
            else related_field.get_column_type()
        )
        related_name = (
            getattr(related_column, "name", None) or related_field.column_name or related_key
        )
        related_nullable = (
            related_column.nullable if related_column is not None else related_field.null
        )
        columns.append(
            sqlalchemy.Column(
                key=self.get_fk_field_name(name, related_key),
                name=self.get_fk_column_name(name, related_name),
                type_=related_type,
                nullable=self.null or related_nullable,
                primary_key=self.primary_key,
                autoincrement=False,
            )
        )
    return columns

def get_global_constraints(
    self,
    name: str,
    columns: typing.Sequence[sqlalchemy.Column],
    *,
    schema: str | None = None,
) -> typing.Sequence[sqlalchemy.Constraint | sqlalchemy.Index]:
    """Build cross-table constraints and indexes for the foreign key columns.

    Args:
        name: Logical field name on the owning model.
        columns: Physical columns generated for the relation.
        schema: Optional schema override for the target table reference.

    Returns:
        typing.Sequence[sqlalchemy.Constraint | sqlalchemy.Index]:
            SQLAlchemy constraints and indexes that should be attached to
            the owning table.
    """
    owner_registry = getattr(self.owner.meta, "registry", None)
    target = self.target
    target_registry = getattr(target.meta, "registry", None)
    owner_database = getattr(self.owner, "database", None)
    target_database = getattr(target, "database", None)
    use_constraint = not (
        self.no_constraint
        or (
            owner_registry not in (None, False)
            and target_registry not in (None, False)
            and owner_registry is not target_registry
        )
        or (
            owner_database is not None
            and target_database is not None
            and owner_database is not target_database
        )
    )
    constraints: list[sqlalchemy.Constraint | sqlalchemy.Index] = []
    if use_constraint:
        table_name = target.meta.tablename
        if schema is not None:
            table_name = f"{schema}.{table_name}"
        constraints.append(
            sqlalchemy.schema.ForeignKeyConstraint(
                columns,
                [
                    f"{table_name}.{self.from_fk_field_name(name, column.key)}"
                    for column in columns
                ],
                ondelete=self.on_delete,
                onupdate=self.on_update,
                name=self.get_fk_name(name),
            )
        )
    if self.unique or self.index:
        constraints.append(
            sqlalchemy.Index(
                self.get_fkindex_name(name),
                *columns,
                unique=self.unique,
            )
        )
    return constraints

def has_column(self) -> bool:
    return not self.is_virtual

def get_embedded_fields(
    self,
    field_name: str,
    existing_fields: dict[str, "Field"],
) -> dict[str, "Field"]:
    del field_name, existing_fields
    return {}

def expand_relationship(self, value: typing.Any) -> typing.Any:
    """Normalize relation values into lightweight target model instances.

    Raw primary-key values, dictionaries, proxy models, and already-loaded
    target instances are all accepted so callers can assign foreign keys in a
    natural way while the ORM preserves a consistent in-memory shape.

    Args:
        value: Incoming relation value.

    Returns:
        typing.Any: Target model instance or `None`.
    """
    if value is None:
        return None
    target = self.target
    related_columns = tuple(self.related_columns.keys())
    if isinstance(value, target):
        if (
            self.null
            and related_columns
            and all(
                key in value.__dict__ and getattr(value, key) is None
                for key in related_columns
            )
        ):
            return None
        return value
    if hasattr(value, "__db_model__"):
        value_cls = value.__class__
        if (
            getattr(value_cls, "is_proxy_model", False)
            and getattr(value_cls, "parent", None) is target
        ):
            if (
                self.null
                and related_columns
                and all(
                    key in value.__dict__ and getattr(value, key) is None
                    for key in related_columns
                )
            ):
                return None
            return value
    if isinstance(value, dict):
        return target(**value)
    if hasattr(value, "pk"):
        pk_value = value.pk
        if isinstance(pk_value, dict):
            return target(pk=pk_value)
        return target(pk=pk_value)
    return target(pk=value)

def clean(
    self, name: str, value: typing.Any, *, for_query: bool = False
) -> dict[str, typing.Any]:
    del for_query
    cleaned: dict[str, typing.Any] = {}
    related_keys = tuple(self.related_columns.keys())
    column_names = self.get_column_names(name)

    if value is None:
        for column_name in column_names:
            cleaned[column_name] = None
        return cleaned

    target = self.target
    if isinstance(value, dict):
        for related_key, column_name in zip(related_keys, column_names, strict=False):
            if related_key in value:
                cleaned[column_name] = value[related_key]
            elif column_name in value:
                cleaned[column_name] = value[column_name]
        return cleaned

    if isinstance(value, target):
        for related_key, column_name in zip(related_keys, column_names, strict=False):
            cleaned[column_name] = getattr(value, related_key, None)
        return cleaned

    if hasattr(value, "__db_model__"):
        value_cls = value.__class__
        if (
            getattr(value_cls, "is_proxy_model", False)
            and getattr(value_cls, "parent", None) is target
        ):
            for related_key, column_name in zip(related_keys, column_names, strict=False):
                cleaned[column_name] = getattr(value, related_key, None)
            return cleaned

    pk_value = getattr(value, "pk", None) if hasattr(value, "pk") else None
    if isinstance(pk_value, dict):
        return self.clean(name, pk_value)
    if pk_value is not None and len(column_names) == 1:
        cleaned[column_names[0]] = pk_value
        return cleaned

    if len(column_names) == 1:
        cleaned[column_names[0]] = value
        return cleaned

    raise ValueError(f"Cannot handle composite foreign key value {value!r}.")

def modify_input(self, name: str, kwargs: dict[str, typing.Any]) -> None:
    if name in kwargs:
        return

    column_names = self.get_column_names(name)
    if len(column_names) <= 1:
        return

    payload: dict[str, typing.Any] = {}
    for column_name in column_names:
        if column_name in kwargs:
            payload[self.from_fk_field_name(name, column_name)] = kwargs.pop(column_name)

    if not payload:
        return
    if len(payload) != len(column_names):
        raise ValueError("Cannot update the foreign key partially")
    kwargs[name] = payload

def raise_for_non_default(self, default: typing.Any, server_default: typing.Any) -> typing.Any:
    del default, server_default

def get_default_value(self) -> typing.Any:
    return self.validator.get_default_value()

def has_default(self) -> bool:
    return self.validator.has_default()

def get_default_values(
    self,
    field_name: str,
    cleaned_data: dict[str, typing.Any],
) -> dict[str, typing.Any]:
    if field_name in cleaned_data or not self.has_column():
        return {}
    return {field_name: self.get_default_value()}

def is_required(self) -> bool:
    if self.primary_key and self.autoincrement:
        return False
    return not (self.null or self.server_default is not None or self.has_default())

def get_is_null_clause(self, column: typing.Any) -> typing.Any:
    return column == None  # noqa: E711

def get_is_empty_clause(self, column: typing.Any) -> typing.Any:
    return self.get_is_null_clause(column)

def operator_to_clause(
    self,
    field_name: str,
    operator: str,
    table: sqlalchemy.Table,
    value: typing.Any,
) -> typing.Any:
    """Translate one lookup operator into a SQLAlchemy clause.

    Args:
        field_name: Logical field name being filtered.
        operator: Saffier lookup suffix such as `exact` or `isnull`.
        table: SQLAlchemy table containing the target column.
        value: Lookup value supplied by the caller.

    Returns:
        Any: SQLAlchemy boolean clause implementing the lookup.
    """
    column = table.columns[field_name]
    mapped_operator = settings.filter_operators.get(operator, operator)

    if mapped_operator == "isnull":
        is_null = self.get_is_null_clause(column)
        return is_null if value else sqlalchemy.not_(is_null)

    if mapped_operator == "isempty":
        is_empty = self.get_is_empty_clause(column)
        return is_empty if value else sqlalchemy.not_(is_empty)

    return getattr(column, mapped_operator)(value)

def get_fk_name(self, name: str) -> str:
    fk_name = f"fk_{self.owner.meta.tablename}_{self.target.meta.tablename}_{name}"
    return fk_name[:CHAR_LIMIT]

def get_fkindex_name(self, name: str) -> str:
    fk_name = f"fkindex_{self.owner.meta.tablename}_{self.target.meta.tablename}_{name}"
    return fk_name[:CHAR_LIMIT]

def get_fk_field_name(self, name: str, fieldname: str) -> str:
    if len(self.related_columns) == 1:
        return name
    return f"{name}_{fieldname}"

def get_fk_column_name(self, name: str, fieldname: str) -> str:
    name = self.column_name or name
    if len(self.related_columns) == 1:
        return name
    return f"{name}_{fieldname}"

def get_column_names(self, name: str) -> tuple[str, ...]:
    return tuple(
        self.get_fk_field_name(name, field_name) for field_name in self.related_columns
    )

def from_fk_field_name(self, name: str, fieldname: str) -> str:
    if len(self.related_columns) == 1:
        return next(iter(self.related_columns.keys()))
    return fieldname.removeprefix(f"{name}_")

async def pre_save_callback(
    self,
    value: typing.Any,
    original_value: typing.Any,
    is_update: bool,
) -> dict[str, typing.Any]:
    """Save nested related objects before persisting the owning row.

    The callback allows callers to pass an unsaved related model instance or
    a dictionary payload. When necessary, the related object is saved first
    and the method returns the concrete foreign-key column values to persist
    on the owning model.

    Args:
        value: Current relation value being persisted.
        original_value: Previous relation value used during updates.
        is_update: Whether the owning model is performing an update.

    Returns:
        dict[str, typing.Any]: Database-ready foreign-key column payload.
    """
    target = self.target
    if value is None or (isinstance(value, dict) and not value):
        value = original_value

    if isinstance(value, target):
        if (
            getattr(value, "_saffier_save_in_progress", False)
            or getattr(value, "pk", None) is not None
        ):
            return self.clean(self.name, value, for_query=False)
        await value.save()
        return self.clean(self.name, value, for_query=False)

    if hasattr(value, "__db_model__"):
        value_cls = value.__class__
        if (
            getattr(value_cls, "is_proxy_model", False)
            and getattr(value_cls, "parent", None) is target
        ):
            if (
                getattr(value, "_saffier_save_in_progress", False)
                or getattr(value, "pk", None) is not None
            ):
                return self.clean(self.name, value, for_query=False)
            await value.save()
            return self.clean(self.name, value, for_query=False)

    if isinstance(value, dict):
        return await self.pre_save_callback(
            target(**value),
            original_value=None,
            is_update=is_update,
        )

    if hasattr(value, "pk") and getattr(value, "pk", None) is not None:
        return self.clean(self.name, value, for_query=False)

    if value is None:
        return {}
    return {self.name: value}

def is_cross_db(self, owner_database: typing.Any | None = None) -> bool:
    if owner_database is None:
        owner_database = getattr(self.owner, "database", None)
        if owner_database is None:
            owner_registry = getattr(getattr(self.owner, "meta", None), "registry", None)
            owner_database = getattr(owner_registry, "database", None)

    target = self.target
    target_database = getattr(target, "database", None)
    if target_database is None:
        target_registry = getattr(getattr(target, "meta", None), "registry", None)
        target_database = getattr(target_registry, "database", None)

    if owner_database is None or target_database is None:
        return False

    return str(owner_database.url) != str(target_database.url)

def get_related_model_for_admin(self) -> typing.Any | None:
    target = self.target
    registry = getattr(getattr(target, "meta", None), "registry", None)
    admin_models = getattr(registry, "admin_models", ())
    if registry and target.__name__ in admin_models:
        return target
    return None

def get_column(self, name: str) -> sqlalchemy.Column:
    return super().get_column(name)