Field¶

def get_column(self, name: str) -> sqlalchemy.Column:
    """Build the SQLAlchemy column used to persist this field.

    Args:
        name: Logical field name declared on the model.

    Returns:
        sqlalchemy.Column: SQLAlchemy column configured from the field
        options and validator metadata.
    """
    column_type = self.get_column_type()
    constraints = self.get_constraints()
    column_kwargs = {
        "primary_key": self.primary_key,
        "nullable": self.get_columns_nullable() and not self.primary_key,
        "index": self.index,
        "unique": self.unique,
        "default": self.default_value,
        "comment": self.comment,
        "server_default": self.server_default,
    }
    if self.autoincrement:
        column_kwargs["autoincrement"] = True

    return sqlalchemy.Column(
        self.column_name or name,
        column_type,
        *constraints,
        key=name,
        **column_kwargs,
    )

def get_validator(self, **kwargs: typing.Any) -> SaffierField:
    return SaffierField(**kwargs)  # pragma: no cover

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]:
    return [self.get_column(name)]

def get_global_constraints(
    self,
    name: str,
    columns: typing.Sequence[sqlalchemy.Column],
    *,
    schema: str | None = None,
) -> typing.Sequence[sqlalchemy.Constraint | sqlalchemy.Index]:
    del name, columns, schema
    return []

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:
    return value

def clean(
    self, name: str, value: typing.Any, *, for_query: bool = False
) -> dict[str, typing.Any]:
    """Normalize one logical field value into column-value pairs.

    Args:
        name: Logical field name.
        value: User-facing field value.
        for_query: Reserved for field implementations that need distinct
            query-time normalization.

    Returns:
        dict[str, Any]: Database payload keyed by column name.
    """
    del for_query
    if not self.has_column():
        return {}
    return {name: value}

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

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)