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_nullable(self) -> bool:
    """Resolve SQLAlchemy nullability for this field during metadata build.

    The model declaration remains the canonical field definition, but
    migration generation can temporarily ask Saffier to render selected
    columns as nullable. That temporary override makes it possible to add a
    required field to a table that already has rows, then backfill defaults
    in the generated online migration before the final model constraint is
    enforced by later migrations.

    Returns:
        bool: ``True`` when the field declaration or active migration
        override should make generated columns nullable.
    """
    if self.null:
        return True

    force_fields = FORCE_FIELDS_NULLABLE.get()
    owner_name = self.owner.__name__ if self.owner is not None else ""
    return (owner_name, self.name) in force_fields or ("", self.name) in force_fields

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.get_columns_nullable() or related_nullable)
                and not self.primary_key,
                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)