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.null 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(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)