1from __future__ import annotations
2
3import sys
4import threading
5from collections import Counter
6from collections.abc import Iterable
7from importlib import import_module
8from importlib.util import find_spec
9
10from plain.exceptions import ImproperlyConfigured, PackageRegistryNotReady
11
12from .config import PackageConfig
13
14_CONFIG_MODULE_NAME = "config"
15
16
17class PackagesRegistry:
18 """
19 A registry that stores the configuration of installed applications.
20
21 It also keeps track of models, e.g. to provide reverse relations.
22 """
23
24 def __init__(self, installed_packages: Iterable[str | PackageConfig] | None = ()):
25 # installed_packages is set to None when creating the main registry
26 # because it cannot be populated at that point. Other registries must
27 # provide a list of installed packages and are populated immediately.
28 if installed_packages is None and hasattr(
29 sys.modules[__name__], "packages_registry"
30 ):
31 raise RuntimeError("You must supply an installed_packages argument.")
32
33 # Mapping of labels to PackageConfig instances for installed packages.
34 self.package_configs: dict[str, PackageConfig] = {}
35
36 # Whether the registry is populated.
37 self.packages_ready = self.ready = False
38
39 # Lock for thread-safe population.
40 self._lock = threading.RLock()
41 self.loading = False
42
43 # Populate packages and models, unless it's the main registry.
44 if installed_packages is not None:
45 self.populate(installed_packages)
46
47 def populate(
48 self, installed_packages: Iterable[str | PackageConfig] | None = None
49 ) -> None:
50 """
51 Load application configurations and models.
52
53 Import each application module and then each model module.
54
55 It is thread-safe and idempotent, but not reentrant.
56 """
57 if self.ready:
58 return
59
60 # populate() might be called by two threads in parallel on servers
61 # that create threads before initializing the request handler.
62 with self._lock:
63 if self.ready:
64 return
65
66 # An RLock prevents other threads from entering this section. The
67 # compare and set operation below is atomic.
68 if self.loading:
69 # Prevent reentrant calls to avoid running PackageConfig.ready()
70 # methods twice.
71 raise RuntimeError("populate() isn't reentrant")
72 self.loading = True
73
74 # Phase 1: initialize app configs and import app modules.
75 if installed_packages is None:
76 return
77
78 for entry in installed_packages:
79 if isinstance(entry, PackageConfig):
80 # Some instances of the registry pass in the
81 # PackageConfig directly...
82 self.register_config(package_config=entry)
83 else:
84 try:
85 import_module(f"{entry}.{_CONFIG_MODULE_NAME}")
86 except ModuleNotFoundError:
87 pass
88
89 # The config for the package should now be registered, if it existed.
90 # And if it didn't, now we can auto generate one.
91 entry_config = None
92 for config in self.package_configs.values():
93 if config.name == entry:
94 entry_config = config
95 break
96
97 if not entry_config:
98 # Use PackageConfig class as-is, without any customization.
99 auto_package_config = PackageConfig(entry)
100 entry_config = self.register_config(auto_package_config)
101
102 # Make sure we have the same number of configs as we have installed packages
103 installed_packages_list = list(installed_packages)
104 if len(self.package_configs) != len(installed_packages_list):
105 raise ImproperlyConfigured(
106 f"The number of installed packages ({len(installed_packages_list)}) does not match the number of "
107 f"registered configs ({len(self.package_configs)})."
108 )
109
110 # Check for duplicate app names.
111 counts = Counter(
112 package_config.name for package_config in self.package_configs.values()
113 )
114 duplicates = [name for name, count in counts.most_common() if count > 1]
115 if duplicates:
116 raise ImproperlyConfigured(
117 "Package names aren't unique, duplicates: {}".format(
118 ", ".join(duplicates)
119 )
120 )
121
122 self.packages_ready = True
123
124 # Phase 3: run ready() methods of app configs.
125 for package_config in self.get_package_configs():
126 package_config.ready()
127
128 self.ready = True
129
130 def check_packages_ready(self) -> None:
131 """Raise an exception if all packages haven't been imported yet."""
132 if not self.packages_ready:
133 from plain.runtime import settings
134
135 # If "not ready" is due to unconfigured settings, accessing
136 # INSTALLED_PACKAGES raises a more helpful ImproperlyConfigured
137 # exception.
138 settings.INSTALLED_PACKAGES
139 raise PackageRegistryNotReady("Packages aren't loaded yet.")
140
141 def get_package_configs(self) -> Iterable[PackageConfig]:
142 """Import applications and return an iterable of app configs."""
143 self.check_packages_ready()
144 return self.package_configs.values()
145
146 def get_package_config(self, package_label: str) -> PackageConfig:
147 """
148 Import applications and returns an app config for the given label.
149
150 Raise LookupError if no application exists with this label.
151 """
152 self.check_packages_ready()
153 try:
154 return self.package_configs[package_label]
155 except KeyError:
156 message = f"No installed app with label '{package_label}'."
157 for package_config in self.get_package_configs():
158 if package_config.name == package_label:
159 message += f" Did you mean '{package_config.package_label}'?"
160 break
161 raise LookupError(message)
162
163 def get_containing_package_config(self, object_name: str) -> PackageConfig | None:
164 """
165 Look for an app config containing a given object.
166
167 object_name is the dotted Python path to the object.
168
169 Return the app config for the inner application in case of nesting.
170 Return None if the object isn't in any registered app config.
171 """
172 self.check_packages_ready()
173 candidates = []
174 for package_config in self.package_configs.values():
175 if object_name.startswith(package_config.name):
176 subpath = object_name.removeprefix(package_config.name)
177 if subpath == "" or subpath[0] == ".":
178 candidates.append(package_config)
179 if candidates:
180 return sorted(candidates, key=lambda ac: -len(ac.name))[0]
181 return None
182
183 def register_config(self, package_config: PackageConfig) -> PackageConfig:
184 """
185 Add a config to the registry.
186
187 Typically used as a decorator on a PackageConfig subclass. Example:
188
189 @register_config
190 class Config(PackageConfig):
191 pass
192 """
193 if package_config.package_label in self.package_configs:
194 raise ImproperlyConfigured(
195 f"Package labels aren't unique, duplicates: {package_config.package_label}"
196 )
197 self.package_configs[package_config.package_label] = package_config
198 package_config.packages = self
199
200 return package_config
201
202 def autodiscover_modules(self, module_name: str, *, include_app: bool) -> None:
203 def _import_if_exists(name: str) -> None:
204 if find_spec(name):
205 import_module(name)
206 return None
207
208 # Load from all packages
209 for package_config in self.get_package_configs():
210 _import_if_exists(f"{package_config.name}.{module_name}")
211
212 # Load from app if requested
213 if include_app:
214 _import_if_exists(f"app.{module_name}")
215
216
217packages_registry = PackagesRegistry(installed_packages=None)
218
219
220def register_config(package_config_class: type[PackageConfig]) -> type[PackageConfig]:
221 """A decorator to register a PackageConfig subclass."""
222 module_name = package_config_class.__module__
223
224 # If it is in .config like expected, return the parent module name
225 if module_name.endswith(f".{_CONFIG_MODULE_NAME}"):
226 module_name = module_name[: -len(_CONFIG_MODULE_NAME) - 1]
227
228 package_config = package_config_class(module_name)
229
230 packages_registry.register_config(package_config)
231
232 return package_config_class