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 pkgutil
2from importlib import import_module
3
4from plain import signals
5from plain.exceptions import ImproperlyConfigured
6from plain.runtime import settings
7from plain.utils.connection import BaseConnectionHandler, ConnectionProxy
8from plain.utils.functional import cached_property
9from plain.utils.module_loading import import_string
10
11DEFAULT_DB_ALIAS = "default"
12PLAIN_VERSION_PICKLE_KEY = "_plain_version"
13
14
15class Error(Exception):
16 pass
17
18
19class InterfaceError(Error):
20 pass
21
22
23class DatabaseError(Error):
24 pass
25
26
27class DataError(DatabaseError):
28 pass
29
30
31class OperationalError(DatabaseError):
32 pass
33
34
35class IntegrityError(DatabaseError):
36 pass
37
38
39class InternalError(DatabaseError):
40 pass
41
42
43class ProgrammingError(DatabaseError):
44 pass
45
46
47class NotSupportedError(DatabaseError):
48 pass
49
50
51class DatabaseErrorWrapper:
52 """
53 Context manager and decorator that reraises backend-specific database
54 exceptions using Plain's common wrappers.
55 """
56
57 def __init__(self, wrapper):
58 """
59 wrapper is a database wrapper.
60
61 It must have a Database attribute defining PEP-249 exceptions.
62 """
63 self.wrapper = wrapper
64
65 def __enter__(self):
66 pass
67
68 def __exit__(self, exc_type, exc_value, traceback):
69 if exc_type is None:
70 return
71 for plain_exc_type in (
72 DataError,
73 OperationalError,
74 IntegrityError,
75 InternalError,
76 ProgrammingError,
77 NotSupportedError,
78 DatabaseError,
79 InterfaceError,
80 Error,
81 ):
82 db_exc_type = getattr(self.wrapper.Database, plain_exc_type.__name__)
83 if issubclass(exc_type, db_exc_type):
84 plain_exc_value = plain_exc_type(*exc_value.args)
85 # Only set the 'errors_occurred' flag for errors that may make
86 # the connection unusable.
87 if plain_exc_type not in (DataError, IntegrityError):
88 self.wrapper.errors_occurred = True
89 raise plain_exc_value.with_traceback(traceback) from exc_value
90
91 def __call__(self, func):
92 # Note that we are intentionally not using @wraps here for performance
93 # reasons. Refs #21109.
94 def inner(*args, **kwargs):
95 with self:
96 return func(*args, **kwargs)
97
98 return inner
99
100
101def load_backend(backend_name):
102 """
103 Return a database backend's "base" module given a fully qualified database
104 backend name, or raise an error if it doesn't exist.
105 """
106 try:
107 return import_module(f"{backend_name}.base")
108 except ImportError as e_user:
109 # The database backend wasn't found. Display a helpful error message
110 # listing all built-in database backends.
111 import plain.models.backends
112
113 builtin_backends = [
114 name
115 for _, name, ispkg in pkgutil.iter_modules(plain.models.backends.__path__)
116 if ispkg and name not in {"base", "dummy"}
117 ]
118 if backend_name not in [f"plain.models.backends.{b}" for b in builtin_backends]:
119 backend_reprs = map(repr, sorted(builtin_backends))
120 raise ImproperlyConfigured(
121 "{!r} isn't an available database backend or couldn't be "
122 "imported. Check the above exception. To use one of the "
123 "built-in backends, use 'plain.models.backends.XXX', where XXX "
124 "is one of:\n"
125 " {}".format(backend_name, ", ".join(backend_reprs))
126 ) from e_user
127 else:
128 # If there's some other error, this must be an error in Plain
129 raise
130
131
132class ConnectionHandler(BaseConnectionHandler):
133 settings_name = "DATABASES"
134
135 def configure_settings(self, databases):
136 databases = super().configure_settings(databases)
137 if databases == {}:
138 databases[DEFAULT_DB_ALIAS] = {"ENGINE": "plain.models.backends.dummy"}
139 elif DEFAULT_DB_ALIAS not in databases:
140 raise ImproperlyConfigured(
141 f"You must define a '{DEFAULT_DB_ALIAS}' database."
142 )
143 elif databases[DEFAULT_DB_ALIAS] == {}:
144 databases[DEFAULT_DB_ALIAS]["ENGINE"] = "plain.models.backends.dummy"
145
146 # Configure default settings.
147 for conn in databases.values():
148 conn.setdefault("AUTOCOMMIT", True)
149 conn.setdefault("ENGINE", "plain.models.backends.dummy")
150 if conn["ENGINE"] == "plain.models.backends." or not conn["ENGINE"]:
151 conn["ENGINE"] = "plain.models.backends.dummy"
152 conn.setdefault("CONN_MAX_AGE", 0)
153 conn.setdefault("CONN_HEALTH_CHECKS", False)
154 conn.setdefault("OPTIONS", {})
155 conn.setdefault("TIME_ZONE", None)
156 for setting in ["NAME", "USER", "PASSWORD", "HOST", "PORT"]:
157 conn.setdefault(setting, "")
158
159 test_settings = conn.setdefault("TEST", {})
160 default_test_settings = [
161 ("CHARSET", None),
162 ("COLLATION", None),
163 ("MIRROR", None),
164 ("NAME", None),
165 ]
166 for key, value in default_test_settings:
167 test_settings.setdefault(key, value)
168 return databases
169
170 @property
171 def databases(self):
172 # Maintained for backward compatibility as some 3rd party packages have
173 # made use of this private API in the past. It is no longer used within
174 # Plain itself.
175 return self.settings
176
177 def create_connection(self, alias):
178 db = self.settings[alias]
179 backend = load_backend(db["ENGINE"])
180 return backend.DatabaseWrapper(db, alias)
181
182
183class ConnectionRouter:
184 def __init__(self, routers=None):
185 """
186 If routers is not specified, default to settings.DATABASE_ROUTERS.
187 """
188 self._routers = routers
189
190 @cached_property
191 def routers(self):
192 if self._routers is None:
193 self._routers = settings.DATABASE_ROUTERS
194 routers = []
195 for r in self._routers:
196 if isinstance(r, str):
197 router = import_string(r)()
198 else:
199 router = r
200 routers.append(router)
201 return routers
202
203 def _router_func(action):
204 def _route_db(self, model, **hints):
205 chosen_db = None
206 for router in self.routers:
207 try:
208 method = getattr(router, action)
209 except AttributeError:
210 # If the router doesn't have a method, skip to the next one.
211 pass
212 else:
213 chosen_db = method(model, **hints)
214 if chosen_db:
215 return chosen_db
216 instance = hints.get("instance")
217 if instance is not None and instance._state.db:
218 return instance._state.db
219 return DEFAULT_DB_ALIAS
220
221 return _route_db
222
223 db_for_read = _router_func("db_for_read")
224 db_for_write = _router_func("db_for_write")
225
226 def allow_relation(self, obj1, obj2, **hints):
227 for router in self.routers:
228 try:
229 method = router.allow_relation
230 except AttributeError:
231 # If the router doesn't have a method, skip to the next one.
232 pass
233 else:
234 allow = method(obj1, obj2, **hints)
235 if allow is not None:
236 return allow
237 return obj1._state.db == obj2._state.db
238
239 def allow_migrate(self, db, package_label, **hints):
240 for router in self.routers:
241 try:
242 method = router.allow_migrate
243 except AttributeError:
244 # If the router doesn't have a method, skip to the next one.
245 continue
246
247 allow = method(db, package_label, **hints)
248
249 if allow is not None:
250 return allow
251 return True
252
253 def allow_migrate_model(self, db, model):
254 return self.allow_migrate(
255 db,
256 model._meta.package_label,
257 model_name=model._meta.model_name,
258 model=model,
259 )
260
261 def get_migratable_models(self, models_registry, package_label, db):
262 """Return app models allowed to be migrated on provided db."""
263 models = models_registry.get_models(package_label=package_label)
264 return [model for model in models if self.allow_migrate_model(db, model)]
265
266
267connections = ConnectionHandler()
268
269router = ConnectionRouter()
270
271# For backwards compatibility. Prefer connections['default'] instead.
272connection = ConnectionProxy(connections, DEFAULT_DB_ALIAS)
273
274
275# Register an event to reset saved queries when a Plain request is started.
276def reset_queries(**kwargs):
277 for conn in connections.all(initialized_only=True):
278 conn.queries_log.clear()
279
280
281signals.request_started.connect(reset_queries)
282
283
284# Register an event to reset transaction state and close connections past
285# their lifetime.
286def close_old_connections(**kwargs):
287 for conn in connections.all(initialized_only=True):
288 conn.close_if_unusable_or_obsolete()
289
290
291signals.request_started.connect(close_old_connections)
292signals.request_finished.connect(close_old_connections)