1from __future__ import annotations
2
3from collections.abc import Sequence
4from typing import TYPE_CHECKING, Any
5
6from plain.models.backends.utils import truncate_name
7from plain.models.db import db_connection
8from plain.packages import packages_registry
9
10if TYPE_CHECKING:
11 from plain.models.base import Model
12 from plain.models.constraints import BaseConstraint
13 from plain.models.indexes import Index
14
15
16class Options:
17 """
18 Model options descriptor and container.
19
20 Acts as both a descriptor (for lazy initialization and access control)
21 and the actual options instance (cached per model class).
22 """
23
24 # Type annotations for attributes set in _create_and_cache
25 # These exist on cached instances, not on the descriptor itself
26 model: type[Model]
27 package_label: str
28 db_table: str
29 db_table_comment: str
30 ordering: Sequence[str]
31 indexes: Sequence[Index]
32 constraints: Sequence[BaseConstraint]
33 required_db_features: Sequence[str]
34 required_db_vendor: str | None
35 _provided_options: set[str]
36
37 def __init__(
38 self,
39 *,
40 db_table: str | None = None,
41 db_table_comment: str | None = None,
42 ordering: Sequence[str] | None = None,
43 indexes: Sequence[Index] | None = None,
44 constraints: Sequence[BaseConstraint] | None = None,
45 required_db_features: Sequence[str] | None = None,
46 required_db_vendor: str | None = None,
47 package_label: str | None = None,
48 ):
49 """
50 Initialize the descriptor with optional configuration.
51
52 This is called ONCE when defining the base Model class, or when
53 a user explicitly sets model_options = Options(...) on their model.
54 The descriptor then creates cached instances per model subclass.
55 """
56 self._config = {
57 "db_table": db_table,
58 "db_table_comment": db_table_comment,
59 "ordering": ordering,
60 "indexes": indexes,
61 "constraints": constraints,
62 "required_db_features": required_db_features,
63 "required_db_vendor": required_db_vendor,
64 "package_label": package_label,
65 }
66 self._cache: dict[type[Model], Options] = {}
67
68 def __get__(self, instance: Any, owner: type[Model]) -> Options:
69 """
70 Descriptor protocol - returns cached Options for the model class.
71
72 This is called when accessing Model.model_options and returns a per-class
73 cached instance created by _create_and_cache().
74
75 Can be accessed from both class and instances:
76 - MyModel.model_options (class access)
77 - my_instance.model_options (instance access - returns class's options)
78 """
79 # Allow instance access - just return the class's options
80 if instance is not None:
81 owner = instance.__class__
82
83 # Skip for the base Model class - return descriptor
84 if owner.__name__ == "Model" and owner.__module__ == "plain.models.base":
85 return self # type: ignore
86
87 # Return cached instance or create new one
88 if owner not in self._cache:
89 return self._create_and_cache(owner)
90
91 return self._cache[owner]
92
93 def _create_and_cache(self, model: type[Model]) -> Options:
94 """Create Options and cache it."""
95 # Create instance without calling __init__
96 instance = Options.__new__(Options)
97
98 # Track which options were explicitly provided by user
99 # Note: package_label is excluded because it's passed separately in migrations
100 instance._provided_options = {
101 k for k, v in self._config.items() if v is not None and k != "package_label"
102 }
103
104 instance.model = model
105
106 # Resolve package_label
107 package_label = self._config.get("package_label")
108 if package_label is None:
109 module = model.__module__
110 package_config = packages_registry.get_containing_package_config(module)
111 if package_config is None:
112 raise RuntimeError(
113 f"Model class {module}.{model.__name__} doesn't declare an explicit "
114 "package_label and isn't in an application in INSTALLED_PACKAGES."
115 )
116 instance.package_label = package_config.package_label
117 else:
118 instance.package_label = package_label
119
120 # Set db_table
121 db_table = self._config.get("db_table")
122 if db_table is None:
123 instance.db_table = truncate_name(
124 f"{instance.package_label}_{model.__name__.lower()}",
125 db_connection.ops.max_name_length(),
126 )
127 else:
128 instance.db_table = db_table
129
130 instance.db_table_comment = self._config.get("db_table_comment") or ""
131 instance.ordering = self._config.get("ordering") or []
132 instance.indexes = self._config.get("indexes") or []
133 instance.constraints = self._config.get("constraints") or []
134 instance.required_db_features = self._config.get("required_db_features") or []
135 instance.required_db_vendor = self._config.get("required_db_vendor")
136
137 # Format names with class interpolation
138 instance.constraints = instance._format_names_with_class(instance.constraints)
139 instance.indexes = instance._format_names_with_class(instance.indexes)
140
141 # Cache early to prevent recursion if needed
142 self._cache[model] = instance
143
144 return instance
145
146 @property
147 def object_name(self) -> str:
148 """The model class name."""
149 return self.model.__name__
150
151 @property
152 def model_name(self) -> str:
153 """The model class name in lowercase."""
154 return self.object_name.lower()
155
156 @property
157 def label(self) -> str:
158 """The model label: package_label.ClassName"""
159 return f"{self.package_label}.{self.object_name}"
160
161 @property
162 def label_lower(self) -> str:
163 """The model label in lowercase: package_label.classname"""
164 return f"{self.package_label}.{self.model_name}"
165
166 def _format_names_with_class(self, objs: list[Any]) -> list[Any]:
167 """Package label/class name interpolation for object names."""
168 new_objs = []
169 for obj in objs:
170 obj = obj.clone()
171 obj.name = obj.name % {
172 "package_label": self.package_label.lower(),
173 "class": self.model.__name__.lower(),
174 }
175 new_objs.append(obj)
176 return new_objs
177
178 def export_for_migrations(self) -> dict[str, Any]:
179 """Export user-provided options for migrations."""
180 options = {}
181 for name in self._provided_options:
182 if name == "indexes":
183 # Clone indexes and ensure names are set
184 indexes = [idx.clone() for idx in self.indexes]
185 for index in indexes:
186 if not index.name:
187 index.set_name_with_model(self.model)
188 options["indexes"] = indexes
189 elif name == "constraints":
190 # Clone constraints
191 options["constraints"] = [con.clone() for con in self.constraints]
192 else:
193 # Use current attribute value
194 options[name] = getattr(self, name)
195 return options
196
197 @property
198 def total_unique_constraints(self) -> list[Any]:
199 """
200 Return a list of total unique constraints. Useful for determining set
201 of fields guaranteed to be unique for all rows.
202 """
203 from plain.models.constraints import UniqueConstraint
204
205 return [
206 constraint
207 for constraint in self.constraints
208 if (
209 isinstance(constraint, UniqueConstraint)
210 and constraint.condition is None
211 and not constraint.contains_expressions
212 )
213 ]
214
215 def can_migrate(self, connection: Any) -> bool:
216 """
217 Return True if the model can/should be migrated on the given
218 `connection` object.
219 """
220 if self.required_db_vendor:
221 return self.required_db_vendor == connection.vendor
222 if self.required_db_features:
223 return all(
224 getattr(connection.features, feat, False)
225 for feat in self.required_db_features
226 )
227 return True
228
229 def __repr__(self) -> str:
230 return f"<Options for {self.model.__name__}>"
231
232 def __str__(self) -> str:
233 return f"{self.package_label}.{self.model.__name__.lower()}"