1"""Describe the data model as the registry sees it.
2
3`describe_models()` reads the model registry and returns one
4`ModelDescription` per model: its fields, its relations with their target,
5on-delete rule and nullability, the relations pointing back at it, and its
6constraints and indexes with their conditions compiled to SQL. Nothing here
7touches the database.
8
9This is what `plain postgres models` prints. It exists so that the facts a
10reader can't reliably derive from a `models.py` by eye (a one-to-one that
11lives in a unique constraint, a partial unique that only looks like one, a
12reverse accessor's name) come from the framework instead. Views built on
13these facts, such as a diagram, belong to whoever reads them.
14"""
15
16from __future__ import annotations
17
18from collections.abc import Sequence
19from dataclasses import asdict, dataclass
20from typing import TYPE_CHECKING, Any
21
22from plain.packages import packages_registry
23
24from .constraints import CheckConstraint, UniqueConstraint
25from .ddl import compile_expression_sql, compile_index_expressions_sql
26from .exceptions import FieldError
27from .fields.primary_key import PrimaryKeyField
28from .fields.related import ForeignKeyField, RelatedField
29from .fields.reverse_descriptors import BaseReverseDescriptor
30from .fields.reverse_related import ManyToManyRel
31from .registry import models_registry
32
33if TYPE_CHECKING:
34 from .base import Model
35 from .query_utils import Q
36
37
38@dataclass
39class FieldDescription:
40 name: str
41 type: str # The field class name, e.g. "TextField"
42 column: str
43 primary_key: bool
44 null: bool
45 # Effective uniqueness: true when an unconditional, single-field unique
46 # constraint covers this field. A partial unique never counts.
47 unique: bool
48
49
50@dataclass
51class RelationDescription:
52 """A forward relation: a foreign key or many-to-many on this model."""
53
54 name: str
55 type: str # "ForeignKeyField" or "ManyToManyField"
56 to: str # Target model label, e.g. "users.User"
57 on_delete: str | None # "CASCADE", "SET_NULL", "RESTRICT"; None for many-to-many
58 null: bool
59 unique: bool # Effective uniqueness, which makes a foreign key one-to-one
60 column: str | None # None for many-to-many
61 through: str | None # The intermediary model label, many-to-many only
62 query_name: str # The name used to filter from the target back to this model
63
64
65@dataclass
66class ReverseDescription:
67 """A relation on another model that points at this one."""
68
69 from_model: str # The pointing model's label
70 field: str # The field on the pointing model
71 type: str # "ForeignKeyField" or "ManyToManyField"
72 one_to_one: bool
73 query_name: str # The name used to filter this model by the pointing one
74 accessor: str | None # A declared ReverseForeignKey/ReverseManyToMany, if any
75
76 def to_dict(self) -> dict[str, Any]:
77 # Written out rather than `asdict()`: the JSON key is "from", which
78 # can't be an attribute name.
79 return {
80 "from": self.from_model,
81 "field": self.field,
82 "type": self.type,
83 "one_to_one": self.one_to_one,
84 "query_name": self.query_name,
85 "accessor": self.accessor,
86 }
87
88
89@dataclass
90class ConstraintDescription:
91 name: str
92 kind: str # "unique" or "check"
93 fields: list[str] # Unique constraints only; empty for expression-based ones
94 expressions: list[str] # Unique constraints over expressions, as SQL
95 condition: str | None # Partial unique: the WHERE clause as SQL
96 check: str | None # Check constraints: the CHECK expression as SQL
97
98
99@dataclass
100class IndexDescription:
101 name: str
102 fields: list[str]
103 expressions: list[str] # Expression indexes, as SQL
104 condition: str | None
105
106
107@dataclass
108class ModelDescription:
109 label: str # "package.ModelName"
110 name: str
111 package: str
112 table: str
113 fields: list[FieldDescription]
114 relations: list[RelationDescription]
115 reverse: list[ReverseDescription]
116 constraints: list[ConstraintDescription]
117 indexes: list[IndexDescription]
118
119 def to_dict(self) -> dict[str, Any]:
120 return {
121 "label": self.label,
122 "name": self.name,
123 "package": self.package,
124 "table": self.table,
125 "fields": [asdict(f) for f in self.fields],
126 "relations": [asdict(r) for r in self.relations],
127 "reverse": [r.to_dict() for r in self.reverse],
128 "constraints": [asdict(c) for c in self.constraints],
129 "indexes": [asdict(i) for i in self.indexes],
130 }
131
132
133def select_models(
134 targets: tuple[str, ...], *, include_packages: bool
135) -> list[type[Model]]:
136 """Pick the models to describe.
137
138 Each target matches a model by its label (`changes.Thread`), its class
139 name (`Thread`), its table (`changes_thread`), or a whole package by
140 label (`changes`), none of them case-sensitive, and the models come back
141 in the order the targets name them. With no targets, every app model is
142 selected in registry order, plus the installed packages' models when
143 `include_packages` is set.
144 """
145 all_models = models_registry.get_models()
146
147 if not targets:
148 if include_packages:
149 return list(all_models)
150 app_package_labels = {
151 package_config.package_label
152 for package_config in packages_registry.get_package_configs()
153 if package_config.is_app
154 }
155 return [
156 model
157 for model in all_models
158 if model.model_options.package_label in app_package_labels
159 ]
160
161 selected: list[type[Model]] = []
162 for target in targets:
163 target_lower = target.lower()
164 matches = [
165 model
166 for model in all_models
167 if model.model_options.label_lower == target_lower
168 or model.__name__.lower() == target_lower
169 or model.model_options.db_table.lower() == target_lower
170 or model.model_options.package_label.lower() == target_lower
171 ]
172 if not matches:
173 raise LookupError(f"No model or package matches {target!r}")
174 for model in matches:
175 if model not in selected:
176 selected.append(model)
177 return selected
178
179
180def describe_models(
181 targets: tuple[str, ...] = (), *, include_packages: bool = False
182) -> list[ModelDescription]:
183 """Describe the selected models (see `select_models`), in that order."""
184 return [
185 describe_model(model)
186 for model in select_models(targets, include_packages=include_packages)
187 ]
188
189
190def describe_model(model: type[Model]) -> ModelDescription:
191 options = model.model_options
192 meta = model._model_meta
193 unique_fields = options.unique_field_names
194
195 fields: list[FieldDescription] = []
196 relations: list[RelationDescription] = []
197 for model_field in meta.fields:
198 if isinstance(model_field, ForeignKeyField):
199 relations.append(
200 RelationDescription(
201 name=model_field.name,
202 type="ForeignKeyField",
203 to=_label_of(model_field.remote_field.model_ref),
204 on_delete=model_field.remote_field.on_delete.name,
205 null=model_field.allow_null,
206 unique=model_field.name in unique_fields,
207 column=model_field.column,
208 through=None,
209 query_name=model_field.related_query_name(),
210 )
211 )
212 else:
213 fields.append(
214 FieldDescription(
215 name=model_field.name,
216 type=type(model_field).__name__,
217 column=model_field.column,
218 primary_key=isinstance(model_field, PrimaryKeyField),
219 null=model_field.allow_null,
220 unique=model_field.name in unique_fields,
221 )
222 )
223
224 for m2m_field in meta.many_to_many:
225 relations.append(
226 RelationDescription(
227 name=m2m_field.name,
228 type="ManyToManyField",
229 to=_label_of(m2m_field.remote_field.model_ref),
230 on_delete=None,
231 null=False,
232 unique=False,
233 column=None,
234 through=_label_of(m2m_field.remote_field.through_ref),
235 query_name=m2m_field.related_query_name(),
236 )
237 )
238
239 declared_accessors = _declared_reverse_accessors(model)
240 reverse: list[ReverseDescription] = []
241 for rel in meta.related_objects:
242 pointing_field = rel.field
243 pointing_model = pointing_field.model
244 is_many_to_many = isinstance(rel, ManyToManyRel)
245 reverse.append(
246 ReverseDescription(
247 from_model=pointing_model.model_options.label,
248 field=pointing_field.name,
249 type="ManyToManyField" if is_many_to_many else "ForeignKeyField",
250 one_to_one=(
251 not is_many_to_many
252 and pointing_field.name
253 in pointing_model.model_options.unique_field_names
254 ),
255 query_name=rel.name,
256 accessor=declared_accessors.get(pointing_field),
257 )
258 )
259
260 constraints: list[ConstraintDescription] = []
261 for constraint in options.constraints:
262 if isinstance(constraint, UniqueConstraint):
263 constraints.append(
264 ConstraintDescription(
265 name=constraint.name,
266 kind="unique",
267 fields=list(constraint.fields),
268 expressions=_compile_expressions(
269 model=model, expressions=constraint.expressions
270 ),
271 condition=_compile_condition(
272 model=model, condition=constraint.condition
273 ),
274 check=None,
275 )
276 )
277 elif isinstance(constraint, CheckConstraint):
278 constraints.append(
279 ConstraintDescription(
280 name=constraint.name,
281 kind="check",
282 fields=[],
283 expressions=[],
284 condition=None,
285 check=_compile_q(model=model, q=constraint.check),
286 )
287 )
288
289 indexes = [
290 IndexDescription(
291 name=index.name,
292 fields=list(index.fields),
293 expressions=_compile_expressions(
294 model=model, expressions=index.expressions
295 ),
296 condition=_compile_condition(model=model, condition=index.condition),
297 )
298 for index in options.indexes
299 ]
300
301 return ModelDescription(
302 label=options.label,
303 name=options.object_name,
304 package=options.package_label,
305 table=options.db_table,
306 fields=fields,
307 relations=relations,
308 reverse=reverse,
309 constraints=constraints,
310 indexes=indexes,
311 )
312
313
314def _label_of(model_ref: str | type[Model]) -> str:
315 """A model's label, or the reference as written when it never resolved.
316
317 A foreign key to a model that doesn't exist (a typo, a package no longer
318 installed) keeps its string reference; the description shows it rather
319 than failing for every model.
320 """
321 if isinstance(model_ref, str):
322 return model_ref
323 return model_ref.model_options.label
324
325
326def _compile_condition(*, model: type[Model], condition: Q | None) -> str | None:
327 """A partial constraint or index condition as SQL, or None when there is none."""
328 if condition is None:
329 return None
330 return _compile_q(model=model, q=condition)
331
332
333def _compile_q(*, model: type[Model], q: Q) -> str:
334 """The expression as SQL, or as written when it can't compile: a
335 condition reaching through a relation that never resolved would otherwise
336 fail the whole listing."""
337 try:
338 return compile_expression_sql(model, q)
339 except FieldError, TypeError:
340 return repr(q)
341
342
343def _compile_expressions(
344 *, model: type[Model], expressions: Sequence[Any]
345) -> list[str]:
346 """Each index or constraint expression as SQL, parenthesized as in its DDL."""
347 return [
348 compile_index_expressions_sql(model, (expression,))
349 for expression in expressions
350 ]
351
352
353def _declared_reverse_accessors(model: type[Model]) -> dict[RelatedField, str]:
354 """Map each pointing field to the ReverseForeignKey/ReverseManyToMany
355 attribute declared on `model` for it.
356
357 Every live descriptor sits in the model's own namespace: model setup
358 copies inherited ones onto the model before contributing them.
359 """
360 accessors: dict[RelatedField, str] = {}
361 for value in vars(model).values():
362 if not isinstance(value, BaseReverseDescriptor):
363 continue
364 if value.name is None or value.reversed_field is None:
365 continue
366 accessors[value.reversed_field] = value.name
367 return accessors