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