plain.models
Model your data and store it in a database.
# app/users/models.py
from plain import models
from plain.passwords.models import PasswordField
class User(models.Model):
email = models.EmailField()
password = PasswordField()
is_admin = models.BooleanField(default=False)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return self.email
Create, update, and delete instances of your models:
from .models import User
# Create a new user
user = User.objects.create(
email="[email protected]",
password="password",
)
# Update a user
user.email = "[email protected]"
user.save()
# Delete a user
user.delete()
# Query for users
admin_users = User.objects.filter(is_admin=True)
Installation
# app/settings.py
INSTALLED_PACKAGES = [
...
"plain.models",
]
To connect to a database, you can provide a DATABASE_URL environment variable.
DATABASE_URL=postgresql://user:password@localhost:5432/dbname
Or you can manually define the DATABASES setting.
# app/settings.py
DATABASES = {
"default": {
"ENGINE": "plain.models.backends.postgresql",
"NAME": "dbname",
"USER": "user",
"PASSWORD": "password",
"HOST": "localhost",
"PORT": "5432",
}
}
Multiple backends are supported, including Postgres, MySQL, and SQLite.
Querying
Migrations
Fields
Validation
Indexes and constraints
Managers
Forms
1import functools
2import warnings
3from collections import defaultdict
4from functools import partial
5
6
7class ModelsRegistryNotReady(Exception):
8 """The plain.models registry is not populated yet"""
9
10 pass
11
12
13class ModelsRegistry:
14 def __init__(self):
15 # Mapping of app labels => model names => model classes. Every time a
16 # model is imported, ModelBase.__new__ calls packages.register_model which
17 # creates an entry in all_models. All imported models are registered,
18 # regardless of whether they're defined in an installed application
19 # and whether the registry has been populated. Since it isn't possible
20 # to reimport a module safely (it could reexecute initialization code)
21 # all_models is never overridden or reset.
22 self.all_models = defaultdict(dict)
23
24 # Maps ("package_label", "modelname") tuples to lists of functions to be
25 # called when the corresponding model is ready. Used by this class's
26 # `lazy_model_operation()` and `do_pending_operations()` methods.
27 self._pending_operations = defaultdict(list)
28
29 self.ready = False
30
31 def check_ready(self):
32 """Raise an exception if all models haven't been imported yet."""
33 if not self.ready:
34 raise ModelsRegistryNotReady("Models aren't loaded yet.")
35
36 # This method is performance-critical at least for Plain's test suite.
37 @functools.cache
38 def get_models(self, *, package_label=""):
39 """
40 Return a list of all installed models.
41
42 By default, the following models aren't included:
43
44 - auto-created models for many-to-many relations without
45 an explicit intermediate table,
46
47 Set the corresponding keyword argument to True to include such models.
48 """
49
50 self.check_ready()
51
52 models = []
53
54 # Get models for a single package
55 if package_label:
56 package_models = self.all_models[package_label]
57 for model in package_models.values():
58 models.append(model)
59 return models
60
61 # Get models for all packages
62 for package_models in self.all_models.values():
63 for model in package_models.values():
64 models.append(model)
65
66 return models
67
68 def get_model(self, package_label, model_name=None, require_ready=True):
69 """
70 Return the model matching the given package_label and model_name.
71
72 As a shortcut, package_label may be in the form <package_label>.<model_name>.
73
74 model_name is case-insensitive.
75
76 Raise LookupError if no application exists with this label, or no
77 model exists with this name in the application. Raise ValueError if
78 called with a single argument that doesn't contain exactly one dot.
79 """
80
81 if require_ready:
82 self.check_ready()
83
84 if model_name is None:
85 package_label, model_name = package_label.split(".")
86
87 package_models = self.all_models[package_label]
88 return package_models[model_name.lower()]
89
90 def register_model(self, package_label, model):
91 # Since this method is called when models are imported, it cannot
92 # perform imports because of the risk of import loops. It mustn't
93 # call get_package_config().
94 model_name = model._meta.model_name
95 app_models = self.all_models[package_label]
96 if model_name in app_models:
97 if (
98 model.__name__ == app_models[model_name].__name__
99 and model.__module__ == app_models[model_name].__module__
100 ):
101 warnings.warn(
102 f"Model '{package_label}.{model_name}' was already registered. Reloading models is not "
103 "advised as it can lead to inconsistencies, most notably with "
104 "related models.",
105 RuntimeWarning,
106 stacklevel=2,
107 )
108 else:
109 raise RuntimeError(
110 f"Conflicting '{model_name}' models in application '{package_label}': {app_models[model_name]} and {model}."
111 )
112 app_models[model_name] = model
113 self.do_pending_operations(model)
114 self.clear_cache()
115
116 def _get_registered_model(self, package_label, model_name):
117 """
118 Similar to get_model(), but doesn't require that an app exists with
119 the given package_label.
120
121 It's safe to call this method at import time, even while the registry
122 is being populated.
123 """
124 model = self.all_models[package_label].get(model_name.lower())
125 if model is None:
126 raise LookupError(f"Model '{package_label}.{model_name}' not registered.")
127 return model
128
129 def clear_cache(self):
130 """
131 Clear all internal caches, for methods that alter the app registry.
132
133 This is mostly used in tests.
134 """
135 # Call expire cache on each model. This will purge
136 # the relation tree and the fields cache.
137 self.get_models.cache_clear()
138 if self.ready:
139 # Circumvent self.get_models() to prevent that the cache is refilled.
140 # This particularly prevents that an empty value is cached while cloning.
141 for package_models in self.all_models.values():
142 for model in package_models.values():
143 model._meta._expire_cache()
144
145 def lazy_model_operation(self, function, *model_keys):
146 """
147 Take a function and a number of ("package_label", "modelname") tuples, and
148 when all the corresponding models have been imported and registered,
149 call the function with the model classes as its arguments.
150
151 The function passed to this method must accept exactly n models as
152 arguments, where n=len(model_keys).
153 """
154 # Base case: no arguments, just execute the function.
155 if not model_keys:
156 function()
157 # Recursive case: take the head of model_keys, wait for the
158 # corresponding model class to be imported and registered, then apply
159 # that argument to the supplied function. Pass the resulting partial
160 # to lazy_model_operation() along with the remaining model args and
161 # repeat until all models are loaded and all arguments are applied.
162 else:
163 next_model, *more_models = model_keys
164
165 # This will be executed after the class corresponding to next_model
166 # has been imported and registered. The `func` attribute provides
167 # duck-type compatibility with partials.
168 def apply_next_model(model):
169 next_function = partial(apply_next_model.func, model)
170 self.lazy_model_operation(next_function, *more_models)
171
172 apply_next_model.func = function
173
174 # If the model has already been imported and registered, partially
175 # apply it to the function now. If not, add it to the list of
176 # pending operations for the model, where it will be executed with
177 # the model class as its sole argument once the model is ready.
178 try:
179 model_class = self._get_registered_model(*next_model)
180 except LookupError:
181 self._pending_operations[next_model].append(apply_next_model)
182 else:
183 apply_next_model(model_class)
184
185 def do_pending_operations(self, model):
186 """
187 Take a newly-prepared model and pass it to each function waiting for
188 it. This is called at the very end of Models.register_model().
189 """
190 key = model._meta.package_label, model._meta.model_name
191 for function in self._pending_operations.pop(key, []):
192 function(model)
193
194
195models_registry = ModelsRegistry()
196
197
198# Decorator to register a model (using the internal registry for the correct state).
199def register_model(model_class):
200 model_class._meta.models_registry.register_model(
201 model_class._meta.package_label, model_class
202 )
203 return model_class