(self)
| 257 | |
| 258 | @property |
| 259 | def pool(self): |
| 260 | if not self.is_pool: |
| 261 | return None |
| 262 | |
| 263 | if self.settings_dict.get("CONN_MAX_AGE", 0) != 0: |
| 264 | raise ImproperlyConfigured( |
| 265 | "Pooling doesn't support persistent connections." |
| 266 | ) |
| 267 | |
| 268 | pool_key = (self.alias, self.settings_dict["USER"]) |
| 269 | if pool_key not in self._connection_pools: |
| 270 | connect_kwargs = self.get_connection_params() |
| 271 | pool_options = connect_kwargs.pop("pool") |
| 272 | if pool_options is not True: |
| 273 | connect_kwargs.update(pool_options) |
| 274 | |
| 275 | pool = Database.create_pool( |
| 276 | user=self.settings_dict["USER"], |
| 277 | password=self.settings_dict["PASSWORD"], |
| 278 | dsn=dsn(self.settings_dict), |
| 279 | **connect_kwargs, |
| 280 | ) |
| 281 | self._connection_pools.setdefault(pool_key, pool) |
| 282 | |
| 283 | return self._connection_pools[pool_key] |
| 284 | |
| 285 | def close_pool(self): |
| 286 | if self.pool: |
nothing calls this directly
no test coverage detected