BaseField class¶

def get_server_default(self) -> Any:
    """
    Retrieves the server-side default value for the column.

    This method determines the appropriate server default, considering
    explicitly defined `server_default`, the field's `default` value,
    and the `auto_compute_server_default` setting.

    Returns:
        The server default value, which can be `None`, a literal value,
        or a SQLAlchemy text construct.
    """
    # Get the explicitly defined server_default.
    server_default = getattr(self, "server_default", Undefined)
    # If an explicit server_default was provided, return it.
    if server_default is not Undefined:
        return server_default

    # Get the Python default value.
    default = self.default
    # If the Python default is Undefined or None, there's no server default to compute.
    if default is Undefined or default is None:
        return None

    # Determine if auto-computation of server default is enabled based on
    # field settings and global configuration.
    if not callable(default):
        if self.auto_compute_server_default is None:
            # If auto_compute_server_default is None, use global settings and nullability.
            auto_compute_server_default: bool = (
                not self.null and settings.allow_auto_compute_server_defaults
            )
        elif self.auto_compute_server_default == "ignore_null":
            # If "ignore_null", use global settings regardless of nullability.
            auto_compute_server_default = settings.allow_auto_compute_server_defaults
        else:
            # Otherwise, use the explicit auto_compute_server_default setting.
            auto_compute_server_default = self.auto_compute_server_default
    else:
        # If the default is a callable, auto_compute_server_default must be explicitly True.
        auto_compute_server_default = self.auto_compute_server_default is True

    # If auto-computation is enabled, customize the default for server use; otherwise, return None.
    if auto_compute_server_default:
        return self.customize_default_for_server_default(default)
    return None

def customize_default_for_server_default(self, default: Any) -> Any:
    """
    Customizes the default value for use as a server-side default.

    This method ensures that callable defaults are executed and that
    the resulting value is wrapped in a `sqlalchemy.text` construct
    for appropriate server-side default handling.

    Args:
        default: The original default Python value, which might be a callable.

    Returns:
        A `sqlalchemy.text` object or a literal value suitable for a
        server-side default.
    """
    # If the default is a callable, execute it to get the actual value.
    if callable(default):
        default = default()
    # Wrap the default value in a SQLAlchemy text construct with a bind parameter.
    return sqlalchemy.text(":value").bindparams(value=default)

def get_columns_nullable(self) -> bool:
    """
    Determines if the database column(s) corresponding to this field should be nullable.

    This method considers the field's explicit `null` setting and global
    `FORCE_FIELDS_NULLABLE` context variable.

    Returns:
        `True` if the column(s) should be nullable, `False` otherwise.
    """
    # If the field is explicitly nullable, return True.
    if self.null:
        return True
    # Get the set of fields that are forced to be nullable globally.
    force_fields = FORCE_FIELDS_NULLABLE.get()
    # Check if the current field (by owner name and field name, or by just field name)
    # is present in the forced nullable fields.
    return (self.owner.__name__, self.name) in force_fields or ("", self.name) in force_fields

def operator_to_clause(
    self, field_name: str, operator: str, table: sqlalchemy.Table, value: Any
) -> Any:
    """
    Converts a given operator and value into a SQLAlchemy clause for filtering.

    This method maps common operation codes to SQLAlchemy's `ColumnElement` methods,
    allowing for dynamic construction of database queries. It handles special cases
    like case-insensitive exact matches ('iexact') and null checks ('isnull'),
    as well as various string containment operations.

    Args:
        field_name: The name of the column to apply the operator to.
        operator: The operation code (e.g., 'iexact', 'contains', 'isnull').
        table: The SQLAlchemy Table object the column belongs to.
        value: The value to compare against.

    Returns:
        A SQLAlchemy clause suitable for use in a query's WHERE statement.

    Raises:
        KeyError: If 'field_name' does not correspond to an existing column in the table.
        AttributeError: If the mapped operator does not exist as a method on the column.
    """
    # Get the SQLAlchemy Column object from the table using the field_name.
    column = table.columns[field_name]
    # Get the mapped operator from the operator_mapping, or use the original operator if not found.
    mapped_operator = self.operator_mapping.get(operator, operator)

    # Handle 'iexact' (case-insensitive exact match) specifically.
    match mapped_operator:
        case "iexact":
            # Define characters that need to be escaped in LIKE/ILIKE patterns.
            escape_characters = ["%", "_"]
            # Check if the value contains any of the escape characters.
            has_escaped_character = any(char in value for char in escape_characters)

            if has_escaped_character:
                # If escaped characters are present, escape backslashes first, then the specific escape characters.
                processed_value = value.replace("\\", "\\\\")
                for char in escape_characters:
                    processed_value = processed_value.replace(char, f"\\{char}")
                # Use ilike with an explicit escape character.
                return column.ilike(processed_value, escape="\\")
            else:
                # If no escape characters, use ilike directly.
                return column.ilike(value)
        case "isnull" | "isempty":
            # Handle 'isnull' (checking for NULL values).
            isnull = column == None  # noqa: E711
            # is_(None) doesn't work for all fields
            # column == None is required for IPAddressField, DateField, DateTimeField
            if value:
                # we use the native way
                return isnull
            else:
                # emulation for special fields
                # NOTE: tested by "tests/models/test_model_isnull.py",
                #       it is not solvable by just using sqla.true()/false()
                return sqlalchemy.not_(isnull)  # type: ignore

        # Handle various string containment and prefix/suffix matching operations.
        case (
            "contains" | "icontains" | "startswith" | "endswith" | "istartswith" | "iendswith"
        ):
            # Apply the column method with autoescape enabled.
            return getattr(column, mapped_operator)(value, autoescape=True)

        case _:
            # For all other operators, directly apply the corresponding method
            # from the SQLAlchemy column object.
            return getattr(column, mapped_operator)(value)

def is_required(self) -> bool:
    """
    Checks if the field is required for model instantiation and database inserts.

    A field is considered not required if it's a primary key with autoincrement,
    is nullable, has a server default, or has a Python default.

    Returns:
        `True` if the field is required, `False` otherwise.
    """
    # If the field is a primary key and is autoincrementing, it's not required.
    if self.primary_key and self.autoincrement:
        return False
    # The field is not required if it is nullable, has a server default (explicitly set),
    # or has a Python default.
    return not (
        self.null
        # Check if server_default is explicitly set and not Undefined.
        or (self.server_default is not None and self.server_default is not Undefined)
        or self.has_default()
    )

def has_default(self) -> bool:
    """
    Checks if the field has a default value set in Python.

    This includes cases where `default` is `None` but explicitly set.

    Returns:
        `True` if a Python default is set, `False` otherwise.
    """
    # A default exists if it's not None (unless explicit_none is True) and it's not Undefined.
    return bool(
        (self.default is not None or self.explicit_none) and self.default is not Undefined
    )

def get_columns(self, name: str) -> Sequence[sqlalchemy.Column]:
    """
    Returns a sequence of SQLAlchemy Column objects associated with this field.

    This is a base implementation that returns an empty list, as `BaseField`
    itself doesn't directly correspond to columns. Subclasses like `Field`
    will override this to return actual columns.

    Args:
        name: The name of the field.

    Returns:
        An empty sequence of SQLAlchemy Column objects.
    """
    return []

def embed_field(
    self,
    prefix: str,
    new_fieldname: str,
    owner: type[BaseModelType] | None = None,
    parent: BaseFieldType | None = None,
) -> BaseField | None:
    """
    Embeds this field into a new context, potentially modifying its name
    and owner.

    This method creates a copy of the field and updates its `name` and `owner`
    attributes. If a `parent` field specifies `prefix_column_name`, it
    updates the `column_name` of the copied field accordingly. Returns `None`
    if embedding is not supported for this field type.

    Args:
        prefix: The prefix to be added to the field's new name.
        new_fieldname: The new full name of the field after embedding.
        owner: The new owner model class for the embedded field.
        parent: The parent field if this field is being embedded within another
                composite field.

    Returns:
        A copy of the field with updated `name` and `owner`, or `None` if
        embedding is not permitted.
    """
    # Create a shallow copy of the field.
    field_copy = copy.copy(self)
    # Update the name of the copied field.
    field_copy.name = new_fieldname
    # Set the new owner for the copied field.
    field_copy.owner = owner
    # If the parent field has a prefix_column_name attribute, apply it.
    if getattr(parent, "prefix_column_name", None):
        # If the copied field already has a column_name, prefix it.
        if getattr(field_copy, "column_name", None):
            field_copy.column_name = f"{parent.prefix_column_name}{field_copy.column_name}"  # type: ignore
        else:
            # Otherwise, construct the new column_name based on the prefix and new_fieldname.
            field_copy.column_name = (
                f"{parent.prefix_column_name}{new_fieldname[len(prefix) :]}"  # type: ignore
            )

    # Return the modified copy of the field.
    return field_copy

def get_default_value(self) -> Any:
    """
    Retrieves the default Python value for a single-valued field.

    If the default is a callable, it is executed to get the value.

    Returns:
        The default value of the field, or `None` if no default is set.
    """
    # Get the default value, defaulting to None if not present.
    default = getattr(self, "default", None)
    # If the default is a callable, call it to get the actual value.
    if callable(default):
        return default()
    # Otherwise, return the default value directly.
    return default

def get_default_values(self, field_name: str, cleaned_data: dict[str, Any]) -> dict[str, Any]:
    """
    Retrieves default values for the field, to be applied during data cleaning.

    For single-column fields, if the field name is not already present in
    `cleaned_data`, its default value is returned. Multi-value fields
    should override this method.

    Args:
        field_name: The name of the field.
        cleaned_data: The dictionary of already cleaned data.

    Returns:
        A dictionary containing the default value for the field if it's not
        already in `cleaned_data`, otherwise an empty dictionary.
    """
    # for multidefaults overwrite in subclasses get_default_values to
    # parse default values differently
    # NOTE: multi value fields should always check here if defaults were already applied
    # NOTE: when build meta fields without columns this should be empty
    # If the field_name is already in cleaned_data, no default is needed.
    if field_name in cleaned_data:
        return {}
    # Return a dictionary with the field's default value.
    return {field_name: self.get_default_value()}

def clean(self, field_name: str, value: Any, for_query: bool = False) -> dict[str, Any]:
    """
    Validates and transforms a Python value into a format suitable for database operations.

    This method is used when preparing data for insertion or update into the database,
    or when building query predicates. Multi-column fields should return a dictionary
    mapping column names to their cleaned values.

    Args:
        field_name (str): The logical name of the field.
        value (Any): The Python value of the field.
        for_query (bool): If `True`, the cleaning is for query construction; otherwise, for saving.

    Returns:
        dict[str, Any]: A dictionary where keys are database column names and values are
                        their cleaned representations.
    """
    return {}

def to_model(
    self,
    field_name: str,
    value: Any,
) -> dict[str, Any]:
    """
    Converts database column values back into the Python representation for the Edgy model.

    This is the inverse of `clean`. It takes raw values retrieved from the database
    and transforms them into the expected Python type for the model instance.

    Args:
        field_name (str): The logical name of the field.
        value (Any): The raw value(s) retrieved from the database.

    Returns:
        dict[str, Any]: A dictionary where keys are field names and values are
                        their Python representations.
    """
    return {field_name: value}

def get_global_constraints(
    self,
    name: str,
    columns: Sequence[sqlalchemy.Column],
    schemes: Sequence[str] = (),
) -> Sequence[sqlalchemy.Constraint | sqlalchemy.Index]:
    """
    Returns global SQLAlchemy constraints or indexes that apply across multiple columns
    or at the table level (e.g., unique constraints involving multiple fields).

    Args:
        name (str): The logical name of the field.
        columns (Sequence[sqlalchemy.Column]): The SQLAlchemy Column objects associated with this field.
        schemes (Sequence[str]): Database schema names (if applicable).

    Returns:
        Sequence[sqlalchemy.Constraint | sqlalchemy.Index]: A sequence of SQLAlchemy constraints or indexes.
    """
    return []

def get_embedded_fields(
    self, field_name: str, fields: dict[str, BaseFieldType]
) -> dict[str, BaseFieldType]:
    """
    Defines and returns additional fields that should be "embedded" or added
    to the model's schema dynamically.

    This is useful for fields that generate related or sub-fields. The returned
    fields should be new instances or copies, and their `owner` should be set.

    Args:
        field_name (str): The logical name of the field.
        fields (dict[str, BaseFieldType]): The existing fields defined on the model.

    Returns:
        dict[str, BaseFieldType]: A dictionary of new field names to `BaseFieldType` instances.
    """
    return {}

def get_column_names(self, name: str = "") -> frozenset[str]:
    """
    Retrieves the set of database column names associated with this field.

    Args:
        name (str): The logical name of the field. If empty, uses `self.name`.

    Returns:
        frozenset[str]: A frozenset of database column names.
    """
    if name:
        return cast("MetaInfo", self.owner.meta).field_to_column_names[name]
    return cast("MetaInfo", self.owner.meta).field_to_column_names[self.name]