Docs
plain-models
plain
models
options.py Raw LLM
import bisect
import inspect
from collections import defaultdict
from functools import cached_property

from plain.exceptions import FieldDoesNotExist
from plain.models import models_registry
from plain.models.constraints import UniqueConstraint
from plain.models.db import db_connection
from plain.models.fields import PrimaryKeyField
from plain.models.query import QuerySet
from plain.utils.datastructures import ImmutableList

PROXY_PARENTS = object()

EMPTY_RELATION_TREE = ()

IMMUTABLE_WARNING = (
    "The return type of '%s' should never be mutated. If you want to manipulate this "
    "list for your own use, make a copy first."
)

DEFAULT_NAMES = (
    "db_table",
    "db_table_comment",
    "queryset_class",
    "ordering",
    "package_label",
    "models_registry",
    "required_db_features",
    "required_db_vendor",
    "indexes",
    "constraints",
)


def make_immutable_fields_list(name, data):
    return ImmutableList(data, warning=IMMUTABLE_WARNING % name)


class Options:
    FORWARD_PROPERTIES = {
        "fields",
        "many_to_many",
        "concrete_fields",
        "local_concrete_fields",
        "_non_pk_concrete_field_names",
        "_forward_fields_map",
        "base_queryset",
        "queryset",
    }
    REVERSE_PROPERTIES = {"related_objects", "fields_map", "_relation_tree"}

    default_models_registry = models_registry

    def __init__(self, meta, package_label=None):
        self._get_fields_cache = {}
        self.local_fields = []
        self.local_many_to_many = []
        self.queryset_class = None
        self.model_name = None
        self.db_table = ""
        self.db_table_comment = ""
        self.ordering = []
        self.indexes = []
        self.constraints = []
        self.object_name = None
        self.package_label = package_label
        self.required_db_features = []
        self.required_db_vendor = None
        self.meta = meta

        # For any non-abstract class, the concrete class is the model
        # in the end of the proxy_for_model chain. In particular, for
        # concrete models, the concrete_model is always the class itself.
        self.concrete_model = None

        # List of all lookups defined in ForeignKey 'limit_choices_to' options
        # from *other* models. Needed for some admin checks. Internal use only.
        self.related_fkey_lookups = []

        # A custom app registry to use, if you're making a separate model set.
        self.models_registry = self.default_models_registry

    @property
    def label(self):
        return f"{self.package_label}.{self.object_name}"

    @property
    def label_lower(self):
        return f"{self.package_label}.{self.model_name}"

    def contribute_to_class(self, cls, name):
        from plain.models.backends.utils import truncate_name

        cls._meta = self
        self.model = cls
        # First, construct the default values for these options.
        self.object_name = cls.__name__
        self.model_name = self.object_name.lower()

        # Store the original user-defined values for each option,
        # for use when serializing the model definition
        self.original_attrs = {}

        # Next, apply any overridden values from 'class Meta'.
        if self.meta:
            meta_attrs = self.meta.__dict__.copy()
            for name in self.meta.__dict__:
                # Ignore any private attributes that Plain doesn't care about.
                # NOTE: We can't modify a dictionary's contents while looping
                # over it, so we loop over the *original* dictionary instead.
                if name.startswith("_"):
                    del meta_attrs[name]
            for attr_name in DEFAULT_NAMES:
                if attr_name in meta_attrs:
                    setattr(self, attr_name, meta_attrs.pop(attr_name))
                    self.original_attrs[attr_name] = getattr(self, attr_name)
                elif hasattr(self.meta, attr_name):
                    setattr(self, attr_name, getattr(self.meta, attr_name))
                    self.original_attrs[attr_name] = getattr(self, attr_name)

            # Package label/class name interpolation for names of constraints and
            # indexes.
            for attr_name in {"constraints", "indexes"}:
                objs = getattr(self, attr_name, [])
                setattr(self, attr_name, self._format_names_with_class(cls, objs))

            # Any leftover attributes must be invalid.
            if meta_attrs != {}:
                raise TypeError(
                    "'class Meta' got invalid attribute(s): {}".format(
                        ",".join(meta_attrs)
                    )
                )

        del self.meta

        # If the db_table wasn't provided, use the package_label + model_name.
        if not self.db_table:
            self.db_table = f"{self.package_label}_{self.model_name}"
            self.db_table = truncate_name(
                self.db_table,
                db_connection.ops.max_name_length(),
            )

    def _format_names_with_class(self, cls, objs):
        """Package label/class name interpolation for object names."""
        new_objs = []
        for obj in objs:
            obj = obj.clone()
            obj.name = obj.name % {
                "package_label": cls._meta.package_label.lower(),
                "class": cls.__name__.lower(),
            }
            new_objs.append(obj)
        return new_objs

    def _prepare(self, model):
        if not any(f.name == "id" for f in self.local_fields):
            model.add_to_class("id", PrimaryKeyField())

    def add_field(self, field, private=False):
        # Insert the given field in the order in which it was created, using
        # the "creation_counter" attribute of the field.
        # Move many-to-many related fields from self.fields into
        # self.many_to_many.
        if field.is_relation and field.many_to_many:
            bisect.insort(self.local_many_to_many, field)
        else:
            bisect.insort(self.local_fields, field)

        # If the field being added is a relation to another known field,
        # expire the cache on this field and the forward cache on the field
        # being referenced, because there will be new relationships in the
        # cache. Otherwise, expire the cache of references *to* this field.
        # The mechanism for getting at the related model is slightly odd -
        # ideally, we'd just ask for field.related_model. However, related_model
        # is a cached property, and all the models haven't been loaded yet, so
        # we need to make sure we don't cache a string reference.
        if (
            field.is_relation
            and hasattr(field.remote_field, "model")
            and field.remote_field.model
        ):
            try:
                field.remote_field.model._meta._expire_cache(forward=False)
            except AttributeError:
                pass
            self._expire_cache()
        else:
            self._expire_cache(reverse=False)

    def __repr__(self):
        return f"<Options for {self.object_name}>"

    def __str__(self):
        return self.label_lower

    def can_migrate(self, connection):
        """
        Return True if the model can/should be migrated on the given
        `connection` object.
        """
        if self.required_db_vendor:
            return self.required_db_vendor == connection.vendor
        if self.required_db_features:
            return all(
                getattr(connection.features, feat, False)
                for feat in self.required_db_features
            )
        return True

    @property
    def base_queryset(self):
        """
        The base queryset is used by Plain's internal operations like cascading
        deletes, migrations, and related object lookups. It provides access to
        all objects in the database without any filtering, ensuring Plain can
        always see the complete dataset when performing framework operations.

        Unlike user-defined querysets which may filter results (e.g. only active
        objects), the base queryset must never filter out rows to prevent
        incomplete results in related queries.
        """
        return QuerySet(model=self.model)

    @property
    def queryset(self):
        if self.queryset_class:
            return self.queryset_class(model=self.model)
        return QuerySet(model=self.model)

    @cached_property
    def fields(self):
        """
        Return a list of all forward fields on the model and its parents,
        excluding ManyToManyFields.

        Private API intended only to be used by Plain itself; get_fields()
        combined with filtering of field properties is the public API for
        obtaining this field list.
        """

        # For legacy reasons, the fields property should only contain forward
        # fields that are not private or with a m2m cardinality. Therefore we
        # pass these three filters as filters to the generator.
        # The third lambda is a longwinded way of checking f.related_model - we don't
        # use that property directly because related_model is a cached property,
        # and all the models may not have been loaded yet; we don't want to cache
        # the string reference to the related_model.
        def is_not_an_m2m_field(f):
            return not (f.is_relation and f.many_to_many)

        def is_not_a_generic_relation(f):
            return not (f.is_relation and f.one_to_many)

        def is_not_a_generic_foreign_key(f):
            return not (
                f.is_relation
                and f.many_to_one
                and not (hasattr(f.remote_field, "model") and f.remote_field.model)
            )

        return make_immutable_fields_list(
            "fields",
            (
                f
                for f in self._get_fields(reverse=False)
                if is_not_an_m2m_field(f)
                and is_not_a_generic_relation(f)
                and is_not_a_generic_foreign_key(f)
            ),
        )

    @cached_property
    def concrete_fields(self):
        """
        Return a list of all concrete fields on the model and its parents.

        Private API intended only to be used by Plain itself; get_fields()
        combined with filtering of field properties is the public API for
        obtaining this field list.
        """
        return make_immutable_fields_list(
            "concrete_fields", (f for f in self.fields if f.concrete)
        )

    @cached_property
    def local_concrete_fields(self):
        """
        Return a list of all concrete fields on the model.

        Private API intended only to be used by Plain itself; get_fields()
        combined with filtering of field properties is the public API for
        obtaining this field list.
        """
        return make_immutable_fields_list(
            "local_concrete_fields", (f for f in self.local_fields if f.concrete)
        )

    @cached_property
    def many_to_many(self):
        """
        Return a list of all many to many fields on the model and its parents.

        Private API intended only to be used by Plain itself; get_fields()
        combined with filtering of field properties is the public API for
        obtaining this list.
        """
        return make_immutable_fields_list(
            "many_to_many",
            (
                f
                for f in self._get_fields(reverse=False)
                if f.is_relation and f.many_to_many
            ),
        )

    @cached_property
    def related_objects(self):
        """
        Return all related objects pointing to the current model. The related
        objects can come from a one-to-one, one-to-many, or many-to-many field
        relation type.

        Private API intended only to be used by Plain itself; get_fields()
        combined with filtering of field properties is the public API for
        obtaining this field list.
        """
        all_related_fields = self._get_fields(
            forward=False, reverse=True, include_hidden=True
        )
        return make_immutable_fields_list(
            "related_objects",
            (
                obj
                for obj in all_related_fields
                if not obj.hidden or obj.field.many_to_many
            ),
        )

    @cached_property
    def _forward_fields_map(self):
        res = {}
        fields = self._get_fields(reverse=False)
        for field in fields:
            res[field.name] = field
            # Due to the way Plain's internals work, get_field() should also
            # be able to fetch a field by attname. In the case of a concrete
            # field with relation, includes the *_id name too
            try:
                res[field.attname] = field
            except AttributeError:
                pass
        return res

    @cached_property
    def fields_map(self):
        res = {}
        fields = self._get_fields(forward=False, include_hidden=True)
        for field in fields:
            res[field.name] = field
            # Due to the way Plain's internals work, get_field() should also
            # be able to fetch a field by attname. In the case of a concrete
            # field with relation, includes the *_id name too
            try:
                res[field.attname] = field
            except AttributeError:
                pass
        return res

    def get_field(self, field_name):
        """
        Return a field instance given the name of a forward or reverse field.
        """
        try:
            # In order to avoid premature loading of the relation tree
            # (expensive) we prefer checking if the field is a forward field.
            return self._forward_fields_map[field_name]
        except KeyError:
            # If the app registry is not ready, reverse fields are
            # unavailable, therefore we throw a FieldDoesNotExist exception.
            if not self.models_registry.ready:
                raise FieldDoesNotExist(
                    f"{self.object_name} has no field named '{field_name}'. The app cache isn't ready yet, "
                    "so if this is an auto-created related field, it won't "
                    "be available yet."
                )

        try:
            # Retrieve field instance by name from cached or just-computed
            # field map.
            return self.fields_map[field_name]
        except KeyError:
            raise FieldDoesNotExist(
                f"{self.object_name} has no field named '{field_name}'"
            )

    def _populate_directed_relation_graph(self):
        """
        This method is used by each model to find its reverse objects. As this
        method is very expensive and is accessed frequently (it looks up every
        field in a model, in every app), it is computed on first access and then
        is set as a property on every model.
        """
        related_objects_graph = defaultdict(list)

        all_models = self.models_registry.get_models()
        for model in all_models:
            opts = model._meta

            fields_with_relations = (
                f
                for f in opts._get_fields(reverse=False)
                if f.is_relation and f.related_model is not None
            )
            for f in fields_with_relations:
                if not isinstance(f.remote_field.model, str):
                    remote_label = f.remote_field.model._meta.concrete_model._meta.label
                    related_objects_graph[remote_label].append(f)

        for model in all_models:
            # Set the relation_tree using the internal __dict__. In this way
            # we avoid calling the cached property. In attribute lookup,
            # __dict__ takes precedence over a data descriptor (such as
            # @cached_property). This means that the _meta._relation_tree is
            # only called if related_objects is not in __dict__.
            related_objects = related_objects_graph[
                model._meta.concrete_model._meta.label
            ]
            model._meta.__dict__["_relation_tree"] = related_objects
        # It seems it is possible that self is not in all_models, so guard
        # against that with default for get().
        return self.__dict__.get("_relation_tree", EMPTY_RELATION_TREE)

    @cached_property
    def _relation_tree(self):
        return self._populate_directed_relation_graph()

    def _expire_cache(self, forward=True, reverse=True):
        # This method is usually called by packages.cache_clear(), when the
        # registry is finalized, or when a new field is added.
        if forward:
            for cache_key in self.FORWARD_PROPERTIES:
                if cache_key in self.__dict__:
                    delattr(self, cache_key)
        if reverse:
            for cache_key in self.REVERSE_PROPERTIES:
                if cache_key in self.__dict__:
                    delattr(self, cache_key)
        self._get_fields_cache = {}

    def get_fields(self, include_hidden=False):
        """
        Return a list of fields associated to the model. By default, include
        forward and reverse fields, fields derived from inheritance, but not
        hidden fields. The returned fields can be changed using the parameters:

        - include_hidden:  include fields that have a related_name that
                           starts with a "+"
        """
        return self._get_fields(include_hidden=include_hidden)

    def _get_fields(
        self,
        forward=True,
        reverse=True,
        include_hidden=False,
        seen_models=None,
    ):
        """
        Internal helper function to return fields of the model.
        * If forward=True, then fields defined on this model are returned.
        * If reverse=True, then relations pointing to this model are returned.
        * If include_hidden=True, then fields with is_hidden=True are returned.
        """

        # This helper function is used to allow recursion in ``get_fields()``
        # implementation and to provide a fast way for Plain's internals to
        # access specific subsets of fields.

        # We must keep track of which models we have already seen. Otherwise we
        # could include the same field multiple times from different models.
        topmost_call = seen_models is None
        if topmost_call:
            seen_models = set()
        seen_models.add(self.model)

        # Creates a cache key composed of all arguments
        cache_key = (forward, reverse, include_hidden, topmost_call)

        try:
            # In order to avoid list manipulation. Always return a shallow copy
            # of the results.
            return self._get_fields_cache[cache_key]
        except KeyError:
            pass

        fields = []

        if reverse:
            # Tree is computed once and cached until the app cache is expired.
            # It is composed of a list of fields pointing to the current model
            # from other models.
            all_fields = self._relation_tree
            for field in all_fields:
                # If hidden fields should be included or the relation is not
                # intentionally hidden, add to the fields dict.
                if include_hidden or not field.remote_field.hidden:
                    fields.append(field.remote_field)

        if forward:
            fields += self.local_fields
            fields += self.local_many_to_many

        # In order to avoid list manipulation. Always
        # return a shallow copy of the results
        fields = make_immutable_fields_list("get_fields()", fields)

        # Store result into cache for later access
        self._get_fields_cache[cache_key] = fields
        return fields

    @cached_property
    def total_unique_constraints(self):
        """
        Return a list of total unique constraints. Useful for determining set
        of fields guaranteed to be unique for all rows.
        """
        return [
            constraint
            for constraint in self.constraints
            if (
                isinstance(constraint, UniqueConstraint)
                and constraint.condition is None
                and not constraint.contains_expressions
            )
        ]

    @cached_property
    def _property_names(self):
        """Return a set of the names of the properties defined on the model."""
        names = []
        for name in dir(self.model):
            attr = inspect.getattr_static(self.model, name)
            if isinstance(attr, property):
                names.append(name)
        return frozenset(names)

    @cached_property
    def _non_pk_concrete_field_names(self):
        """
        Return a set of the non-primary key concrete field names defined on the model.
        """
        names = []
        for field in self.concrete_fields:
            if not field.primary_key:
                names.append(field.name)
                if field.name != field.attname:
                    names.append(field.attname)
        return frozenset(names)

    @cached_property
    def db_returning_fields(self):
        """
        Private API intended only to be used by Plain itself.
        Fields to be returned after a database insert.
        """
        return [
            field
            for field in self._get_fields(forward=True, reverse=False)
            if getattr(field, "db_returning", False)
        ]
On this page