codekingpro/portable-devtools

1from __future__ import annotations2 3import os4import types5import typing as t6import warnings7from weakref import WeakKeyDictionary8 9import sqlalchemy as sa10import sqlalchemy.event as sa_event11import sqlalchemy.exc as sa_exc12import sqlalchemy.orm as sa_orm13from flask import abort14from flask import current_app15from flask import Flask16from flask import has_app_context17 18from .model import _QueryProperty19from .model import BindMixin20from .model import DefaultMeta21from .model import DefaultMetaNoName22from .model import Model23from .model import NameMixin24from .pagination import Pagination25from .pagination import SelectPagination26from .query import Query27from .session import _app_ctx_id28from .session import Session29from .table import _Table30 31_O = t.TypeVar("_O", bound=object)  # Based on sqlalchemy.orm._typing.py32 33 34# Type accepted for model_class argument35_FSA_MCT = t.TypeVar(36    "_FSA_MCT",37    bound=t.Union[38        t.Type[Model],39        sa_orm.DeclarativeMeta,40        t.Type[sa_orm.DeclarativeBase],41        t.Type[sa_orm.DeclarativeBaseNoMeta],42    ],43)44 45 46# Type returned by make_declarative_base47class _FSAModel(Model):48    metadata: sa.MetaData49 50 51def _get_2x_declarative_bases(52    model_class: _FSA_MCT,53) -> list[t.Type[t.Union[sa_orm.DeclarativeBase, sa_orm.DeclarativeBaseNoMeta]]]:54    return [55        b56        for b in model_class.__bases__57        if issubclass(b, (sa_orm.DeclarativeBase, sa_orm.DeclarativeBaseNoMeta))58    ]59 60 61class SQLAlchemy:62    """Integrates SQLAlchemy with Flask. This handles setting up one or more engines,63    associating tables and models with specific engines, and cleaning up connections and64    sessions after each request.65 66    Only the engine configuration is specific to each application, other things like67    the model, table, metadata, and session are shared for all applications using that68    extension instance. Call :meth:`init_app` to configure the extension on an69    application.70 71    After creating the extension, create model classes by subclassing :attr:`Model`, and72    table classes with :attr:`Table`. These can be accessed before :meth:`init_app` is73    called, making it possible to define the models separately from the application.74 75    Accessing :attr:`session` and :attr:`engine` requires an active Flask application76    context. This includes methods like :meth:`create_all` which use the engine.77 78    This class also provides access to names in SQLAlchemy's ``sqlalchemy`` and79    ``sqlalchemy.orm`` modules. For example, you can use ``db.Column`` and80    ``db.relationship`` instead of importing ``sqlalchemy.Column`` and81    ``sqlalchemy.orm.relationship``. This can be convenient when defining models.82 83    :param app: Call :meth:`init_app` on this Flask application now.84    :param metadata: Use this as the default :class:`sqlalchemy.schema.MetaData`. Useful85        for setting a naming convention.86    :param session_options: Arguments used by :attr:`session` to create each session87        instance. A ``scopefunc`` key will be passed to the scoped session, not the88        session instance. See :class:`sqlalchemy.orm.sessionmaker` for a list of89        arguments.90    :param query_class: Use this as the default query class for models and dynamic91        relationships. The query interface is considered legacy in SQLAlchemy.92    :param model_class: Use this as the model base class when creating the declarative93        model class :attr:`Model`. Can also be a fully created declarative model class94        for further customization.95    :param engine_options: Default arguments used when creating every engine. These are96        lower precedence than application config. See :func:`sqlalchemy.create_engine`97        for a list of arguments.98    :param add_models_to_shell: Add the ``db`` instance and all model classes to99        ``flask shell``.100 101    .. versionchanged:: 3.1.0102        The ``metadata`` parameter can still be used with SQLAlchemy 1.x classes,103        but is ignored when using SQLAlchemy 2.x style of declarative classes.104        Instead, specify metadata on your Base class.105 106    .. versionchanged:: 3.1.0107        Added the ``disable_autonaming`` parameter.108 109    .. versionchanged:: 3.1.0110        Changed ``model_class`` parameter to accepta SQLAlchemy 2.x111        declarative base subclass.112 113    .. versionchanged:: 3.0114        An active Flask application context is always required to access ``session`` and115        ``engine``.116 117    .. versionchanged:: 3.0118        Separate ``metadata`` are used for each bind key.119 120    .. versionchanged:: 3.0121        The ``engine_options`` parameter is applied as defaults before per-engine122        configuration.123 124    .. versionchanged:: 3.0125        The session class can be customized in ``session_options``.126 127    .. versionchanged:: 3.0128        Added the ``add_models_to_shell`` parameter.129 130    .. versionchanged:: 3.0131        Engines are created when calling ``init_app`` rather than the first time they132        are accessed.133 134    .. versionchanged:: 3.0135        All parameters except ``app`` are keyword-only.136 137    .. versionchanged:: 3.0138        The extension instance is stored directly as ``app.extensions["sqlalchemy"]``.139 140    .. versionchanged:: 3.0141        Setup methods are renamed with a leading underscore. They are considered142        internal interfaces which may change at any time.143 144    .. versionchanged:: 3.0145        Removed the ``use_native_unicode`` parameter and config.146 147    .. versionchanged:: 2.4148        Added the ``engine_options`` parameter.149 150    .. versionchanged:: 2.1151        Added the ``metadata``, ``query_class``, and ``model_class`` parameters.152 153    .. versionchanged:: 2.1154        Use the same query class across ``session``, ``Model.query`` and155        ``Query``.156 157    .. versionchanged:: 0.16158        ``scopefunc`` is accepted in ``session_options``.159 160    .. versionchanged:: 0.10161        Added the ``session_options`` parameter.162    """163 164    def __init__(165        self,166        app: Flask | None = None,167        *,168        metadata: sa.MetaData | None = None,169        session_options: dict[str, t.Any] | None = None,170        query_class: type[Query] = Query,171        model_class: _FSA_MCT = Model,  # type: ignore[assignment]172        engine_options: dict[str, t.Any] | None = None,173        add_models_to_shell: bool = True,174        disable_autonaming: bool = False,175    ):176        if session_options is None:177            session_options = {}178 179        self.Query = query_class180        """The default query class used by ``Model.query`` and ``lazy="dynamic"``181        relationships.182 183        .. warning::184            The query interface is considered legacy in SQLAlchemy.185 186        Customize this by passing the ``query_class`` parameter to the extension.187        """188 189        self.session = self._make_scoped_session(session_options)190        """A :class:`sqlalchemy.orm.scoping.scoped_session` that creates instances of191        :class:`.Session` scoped to the current Flask application context. The session192        will be removed, returning the engine connection to the pool, when the193        application context exits.194 195        Customize this by passing ``session_options`` to the extension.196 197        This requires that a Flask application context is active.198 199        .. versionchanged:: 3.0200            The session is scoped to the current app context.201        """202 203        self.metadatas: dict[str | None, sa.MetaData] = {}204        """Map of bind keys to :class:`sqlalchemy.schema.MetaData` instances. The205        ``None`` key refers to the default metadata, and is available as206        :attr:`metadata`.207 208        Customize the default metadata by passing the ``metadata`` parameter to the209        extension. This can be used to set a naming convention. When metadata for210        another bind key is created, it copies the default's naming convention.211 212        .. versionadded:: 3.0213        """214 215        if metadata is not None:216            if len(_get_2x_declarative_bases(model_class)) > 0:217                warnings.warn(218                    "When using SQLAlchemy 2.x style of declarative classes,"219                    " the `metadata` should be an attribute of the base class."220                    "The metadata passed into SQLAlchemy() is ignored.",221                    DeprecationWarning,222                    stacklevel=2,223                )224            else:225                metadata.info["bind_key"] = None226                self.metadatas[None] = metadata227 228        self.Table = self._make_table_class()229        """A :class:`sqlalchemy.schema.Table` class that chooses a metadata230        automatically.231 232        Unlike the base ``Table``, the ``metadata`` argument is not required. If it is233        not given, it is selected based on the ``bind_key`` argument.234 235        :param bind_key: Used to select a different metadata.236        :param args: Arguments passed to the base class. These are typically the table's237            name, columns, and constraints.238        :param kwargs: Arguments passed to the base class.239 240        .. versionchanged:: 3.0241            This is a subclass of SQLAlchemy's ``Table`` rather than a function.242        """243 244        self.Model = self._make_declarative_base(245            model_class, disable_autonaming=disable_autonaming246        )247        """A SQLAlchemy declarative model class. Subclass this to define database248        models.249 250        If a model does not set ``__tablename__``, it will be generated by converting251        the class name from ``CamelCase`` to ``snake_case``. It will not be generated252        if the model looks like it uses single-table inheritance.253 254        If a model or parent class sets ``__bind_key__``, it will use that metadata and255        database engine. Otherwise, it will use the default :attr:`metadata` and256        :attr:`engine`. This is ignored if the model sets ``metadata`` or ``__table__``.257 258        For code using the SQLAlchemy 1.x API, customize this model by subclassing259        :class:`.Model` and passing the ``model_class`` parameter to the extension.260        A fully created declarative model class can be261        passed as well, to use a custom metaclass.262 263        For code using the SQLAlchemy 2.x API, customize this model by subclassing264        :class:`sqlalchemy.orm.DeclarativeBase` or265        :class:`sqlalchemy.orm.DeclarativeBaseNoMeta`266        and passing the ``model_class`` parameter to the extension.267        """268 269        if engine_options is None:270            engine_options = {}271 272        self._engine_options = engine_options273        self._app_engines: WeakKeyDictionary[Flask, dict[str | None, sa.engine.Engine]]274        self._app_engines = WeakKeyDictionary()275        self._add_models_to_shell = add_models_to_shell276 277        if app is not None:278            self.init_app(app)279 280    def __repr__(self) -> str:281        if not has_app_context():282            return f"<{type(self).__name__}>"283 284        message = f"{type(self).__name__} {self.engine.url}"285 286        if len(self.engines) > 1:287            message = f"{message} +{len(self.engines) - 1}"288 289        return f"<{message}>"290 291    def init_app(self, app: Flask) -> None:292        """Initialize a Flask application for use with this extension instance. This293        must be called before accessing the database engine or session with the app.294 295        This sets default configuration values, then configures the extension on the296        application and creates the engines for each bind key. Therefore, this must be297        called after the application has been configured. Changes to application config298        after this call will not be reflected.299 300        The following keys from ``app.config`` are used:301 302        - :data:`.SQLALCHEMY_DATABASE_URI`303        - :data:`.SQLALCHEMY_ENGINE_OPTIONS`304        - :data:`.SQLALCHEMY_ECHO`305        - :data:`.SQLALCHEMY_BINDS`306        - :data:`.SQLALCHEMY_RECORD_QUERIES`307        - :data:`.SQLALCHEMY_TRACK_MODIFICATIONS`308 309        :param app: The Flask application to initialize.310        """311        if "sqlalchemy" in app.extensions:312            raise RuntimeError(313                "A 'SQLAlchemy' instance has already been registered on this Flask app."314                " Import and use that instance instead."315            )316 317        app.extensions["sqlalchemy"] = self318        app.teardown_appcontext(self._teardown_session)319 320        if self._add_models_to_shell:321            from .cli import add_models_to_shell322 323            app.shell_context_processor(add_models_to_shell)324 325        basic_uri: str | sa.engine.URL | None = app.config.setdefault(326            "SQLALCHEMY_DATABASE_URI", None327        )328        basic_engine_options = self._engine_options.copy()329        basic_engine_options.update(330            app.config.setdefault("SQLALCHEMY_ENGINE_OPTIONS", {})331        )332        echo: bool = app.config.setdefault("SQLALCHEMY_ECHO", False)333        config_binds: dict[334            str | None, str | sa.engine.URL | dict[str, t.Any]335        ] = app.config.setdefault("SQLALCHEMY_BINDS", {})336        engine_options: dict[str | None, dict[str, t.Any]] = {}337 338        # Build the engine config for each bind key.339        for key, value in config_binds.items():340            engine_options[key] = self._engine_options.copy()341 342            if isinstance(value, (str, sa.engine.URL)):343                engine_options[key]["url"] = value344            else:345                engine_options[key].update(value)346 347        # Build the engine config for the default bind key.348        if basic_uri is not None:349            basic_engine_options["url"] = basic_uri350 351        if "url" in basic_engine_options:352            engine_options.setdefault(None, {}).update(basic_engine_options)353 354        if not engine_options:355            raise RuntimeError(356                "Either 'SQLALCHEMY_DATABASE_URI' or 'SQLALCHEMY_BINDS' must be set."357            )358 359        engines = self._app_engines.setdefault(app, {})360 361        # Dispose existing engines in case init_app is called again.362        if engines:363            for engine in engines.values():364                engine.dispose()365 366            engines.clear()367 368        # Create the metadata and engine for each bind key.369        for key, options in engine_options.items():370            self._make_metadata(key)371            options.setdefault("echo", echo)372            options.setdefault("echo_pool", echo)373            self._apply_driver_defaults(options, app)374            engines[key] = self._make_engine(key, options, app)375 376        if app.config.setdefault("SQLALCHEMY_RECORD_QUERIES", False):377            from . import record_queries378 379            for engine in engines.values():380                record_queries._listen(engine)381 382        if app.config.setdefault("SQLALCHEMY_TRACK_MODIFICATIONS", False):383            from . import track_modifications384 385            track_modifications._listen(self.session)386 387    def _make_scoped_session(388        self, options: dict[str, t.Any]389    ) -> sa_orm.scoped_session[Session]:390        """Create a :class:`sqlalchemy.orm.scoping.scoped_session` around the factory391        from :meth:`_make_session_factory`. The result is available as :attr:`session`.392 393        The scope function can be customized using the ``scopefunc`` key in the394        ``session_options`` parameter to the extension. By default it uses the current395        thread or greenlet id.396 397        This method is used for internal setup. Its signature may change at any time.398 399        :meta private:400 401        :param options: The ``session_options`` parameter from ``__init__``. Keyword402            arguments passed to the session factory. A ``scopefunc`` key is popped.403 404        .. versionchanged:: 3.0405            The session is scoped to the current app context.406 407        .. versionchanged:: 3.0408            Renamed from ``create_scoped_session``, this method is internal.409        """410        scope = options.pop("scopefunc", _app_ctx_id)411        factory = self._make_session_factory(options)412        return sa_orm.scoped_session(factory, scope)413 414    def _make_session_factory(415        self, options: dict[str, t.Any]416    ) -> sa_orm.sessionmaker[Session]:417        """Create the SQLAlchemy :class:`sqlalchemy.orm.sessionmaker` used by418        :meth:`_make_scoped_session`.419 420        To customize, pass the ``session_options`` parameter to :class:`SQLAlchemy`. To421        customize the session class, subclass :class:`.Session` and pass it as the422        ``class_`` key.423 424        This method is used for internal setup. Its signature may change at any time.425 426        :meta private:427 428        :param options: The ``session_options`` parameter from ``__init__``. Keyword429            arguments passed to the session factory.430 431        .. versionchanged:: 3.0432            The session class can be customized.433 434        .. versionchanged:: 3.0435            Renamed from ``create_session``, this method is internal.436        """437        options.setdefault("class_", Session)438        options.setdefault("query_cls", self.Query)439        return sa_orm.sessionmaker(db=self, **options)440 441    def _teardown_session(self, exc: BaseException | None) -> None:442        """Remove the current session at the end of the request.443 444        :meta private:445 446        .. versionadded:: 3.0447        """448        self.session.remove()449 450    def _make_metadata(self, bind_key: str | None) -> sa.MetaData:451        """Get or create a :class:`sqlalchemy.schema.MetaData` for the given bind key.452 453        This method is used for internal setup. Its signature may change at any time.454 455        :meta private:456 457        :param bind_key: The name of the metadata being created.458 459        .. versionadded:: 3.0460        """461        if bind_key in self.metadatas:462            return self.metadatas[bind_key]463 464        if bind_key is not None:465            # Copy the naming convention from the default metadata.466            naming_convention = self._make_metadata(None).naming_convention467        else:468            naming_convention = None469 470        # Set the bind key in info to be used by session.get_bind.471        metadata = sa.MetaData(472            naming_convention=naming_convention, info={"bind_key": bind_key}473        )474        self.metadatas[bind_key] = metadata475        return metadata476 477    def _make_table_class(self) -> type[_Table]:478        """Create a SQLAlchemy :class:`sqlalchemy.schema.Table` class that chooses a479        metadata automatically based on the ``bind_key``. The result is available as480        :attr:`Table`.481 482        This method is used for internal setup. Its signature may change at any time.483 484        :meta private:485 486        .. versionadded:: 3.0487        """488 489        class Table(_Table):490            def __new__(491                cls, *args: t.Any, bind_key: str | None = None, **kwargs: t.Any492            ) -> Table:493                # If a metadata arg is passed, go directly to the base Table. Also do494                # this for no args so the correct error is shown.495                if not args or (len(args) >= 2 and isinstance(args[1], sa.MetaData)):496                    return super().__new__(cls, *args, **kwargs)497 498                metadata = self._make_metadata(bind_key)499                return super().__new__(cls, *[args[0], metadata, *args[1:]], **kwargs)500 501        return Table502 503    def _make_declarative_base(504        self,505        model_class: _FSA_MCT,506        disable_autonaming: bool = False,507    ) -> t.Type[_FSAModel]:508        """Create a SQLAlchemy declarative model class. The result is available as509        :attr:`Model`.510 511        To customize, subclass :class:`.Model` and pass it as ``model_class`` to512        :class:`SQLAlchemy`. To customize at the metaclass level, pass an already513        created declarative model class as ``model_class``.514 515        This method is used for internal setup. Its signature may change at any time.516 517        :meta private:518 519        :param model_class: A model base class, or an already created declarative model520        class.521 522        :param disable_autonaming: Turns off automatic tablename generation in models.523 524        .. versionchanged:: 3.1.0525            Added support for passing SQLAlchemy 2.x base class as model class.526            Added optional ``disable_autonaming`` parameter.527 528        .. versionchanged:: 3.0529            Renamed with a leading underscore, this method is internal.530 531        .. versionchanged:: 2.3532            ``model`` can be an already created declarative model class.533        """534        model: t.Type[_FSAModel]535        declarative_bases = _get_2x_declarative_bases(model_class)536        if len(declarative_bases) > 1:537            # raise error if more than one declarative base is found538            raise ValueError(539                "Only one declarative base can be passed to SQLAlchemy."540                " Got: {}".format(model_class.__bases__)541            )542        elif len(declarative_bases) == 1:543            body = dict(model_class.__dict__)544            body["__fsa__"] = self545            mixin_classes = [BindMixin, NameMixin, Model]546            if disable_autonaming:547                mixin_classes.remove(NameMixin)548            model = types.new_class(549                "FlaskSQLAlchemyBase",550                (*mixin_classes, *model_class.__bases__),551                {"metaclass": type(declarative_bases[0])},552                lambda ns: ns.update(body),553            )554        elif not isinstance(model_class, sa_orm.DeclarativeMeta):555            metadata = self._make_metadata(None)556            metaclass = DefaultMetaNoName if disable_autonaming else DefaultMeta557            model = sa_orm.declarative_base(558                metadata=metadata, cls=model_class, name="Model", metaclass=metaclass559            )560        else:561            model = model_class  # type: ignore[assignment]562 563        if None not in self.metadatas:564            # Use the model's metadata as the default metadata.565            model.metadata.info["bind_key"] = None566            self.metadatas[None] = model.metadata567        else:568            # Use the passed in default metadata as the model's metadata.569            model.metadata = self.metadatas[None]570 571        model.query_class = self.Query572        model.query = _QueryProperty()  # type: ignore[assignment]573        model.__fsa__ = self574        return model575 576    def _apply_driver_defaults(self, options: dict[str, t.Any], app: Flask) -> None:577        """Apply driver-specific configuration to an engine.578 579        SQLite in-memory databases use ``StaticPool`` and disable ``check_same_thread``.580        File paths are relative to the app's :attr:`~flask.Flask.instance_path`,581        which is created if it doesn't exist.582 583        MySQL sets ``charset="utf8mb4"``, and ``pool_timeout`` defaults to 2 hours.584 585        This method is used for internal setup. Its signature may change at any time.586 587        :meta private:588 589        :param options: Arguments passed to the engine.590        :param app: The application that the engine configuration belongs to.591 592        .. versionchanged:: 3.0593            SQLite paths are relative to ``app.instance_path``. It does not use594            ``NullPool`` if ``pool_size`` is 0. Driver-level URIs are supported.595 596        .. versionchanged:: 3.0597            MySQL sets ``charset="utf8mb4". It does not set ``pool_size`` to 10. It598            does not set ``pool_recycle`` if not using a queue pool.599 600        .. versionchanged:: 3.0601            Renamed from ``apply_driver_hacks``, this method is internal. It does not602            return anything.603 604        .. versionchanged:: 2.5605            Returns ``(sa_url, options)``.606        """607        url = sa.engine.make_url(options["url"])608 609        if url.drivername in {"sqlite", "sqlite+pysqlite"}:610            if url.database is None or url.database in {"", ":memory:"}:611                options["poolclass"] = sa.pool.StaticPool612 613                if "connect_args" not in options:614                    options["connect_args"] = {}615 616                options["connect_args"]["check_same_thread"] = False617            else:618                # the url might look like sqlite:///file:path?uri=true619                is_uri = url.query.get("uri", False)620 621                if is_uri:622                    db_str = url.database[5:]623                else:624                    db_str = url.database625 626                if not os.path.isabs(db_str):627                    os.makedirs(app.instance_path, exist_ok=True)628                    db_str = os.path.join(app.instance_path, db_str)629 630                    if is_uri:631                        db_str = f"file:{db_str}"632 633                    options["url"] = url.set(database=db_str)634        elif url.drivername.startswith("mysql"):635            # set queue defaults only when using queue pool636            if (637                "pool_class" not in options638                or options["pool_class"] is sa.pool.QueuePool639            ):640                options.setdefault("pool_recycle", 7200)641 642            if "charset" not in url.query:643                options["url"] = url.update_query_dict({"charset": "utf8mb4"})644 645    def _make_engine(646        self, bind_key: str | None, options: dict[str, t.Any], app: Flask647    ) -> sa.engine.Engine:648        """Create the :class:`sqlalchemy.engine.Engine` for the given bind key and app.649 650        To customize, use :data:`.SQLALCHEMY_ENGINE_OPTIONS` or651        :data:`.SQLALCHEMY_BINDS` config. Pass ``engine_options`` to :class:`SQLAlchemy`652        to set defaults for all engines.653 654        This method is used for internal setup. Its signature may change at any time.655 656        :meta private:657 658        :param bind_key: The name of the engine being created.659        :param options: Arguments passed to the engine.660        :param app: The application that the engine configuration belongs to.661 662        .. versionchanged:: 3.0663            Renamed from ``create_engine``, this method is internal.664        """665        return sa.engine_from_config(options, prefix="")666 667    @property668    def metadata(self) -> sa.MetaData:669        """The default metadata used by :attr:`Model` and :attr:`Table` if no bind key670        is set.671        """672        return self.metadatas[None]673 674    @property675    def engines(self) -> t.Mapping[str | None, sa.engine.Engine]:676        """Map of bind keys to :class:`sqlalchemy.engine.Engine` instances for current677        application. The ``None`` key refers to the default engine, and is available as678        :attr:`engine`.679 680        To customize, set the :data:`.SQLALCHEMY_BINDS` config, and set defaults by681        passing the ``engine_options`` parameter to the extension.682 683        This requires that a Flask application context is active.684 685        .. versionadded:: 3.0686        """687        app = current_app._get_current_object()  # type: ignore[attr-defined]688 689        if app not in self._app_engines:690            raise RuntimeError(691                "The current Flask app is not registered with this 'SQLAlchemy'"692                " instance. Did you forget to call 'init_app', or did you create"693                " multiple 'SQLAlchemy' instances?"694            )695 696        return self._app_engines[app]697 698    @property699    def engine(self) -> sa.engine.Engine:700        """The default :class:`~sqlalchemy.engine.Engine` for the current application,701        used by :attr:`session` if the :attr:`Model` or :attr:`Table` being queried does702        not set a bind key.703 704        To customize, set the :data:`.SQLALCHEMY_ENGINE_OPTIONS` config, and set705        defaults by passing the ``engine_options`` parameter to the extension.706 707        This requires that a Flask application context is active.708        """709        return self.engines[None]710 711    def get_engine(712        self, bind_key: str | None = None, **kwargs: t.Any713    ) -> sa.engine.Engine:714        """Get the engine for the given bind key for the current application.715        This requires that a Flask application context is active.716 717        :param bind_key: The name of the engine.718 719        .. deprecated:: 3.0720            Will be removed in Flask-SQLAlchemy 3.2. Use ``engines[key]`` instead.721 722        .. versionchanged:: 3.0723            Renamed the ``bind`` parameter to ``bind_key``. Removed the ``app``724            parameter.725        """726        warnings.warn(727            "'get_engine' is deprecated and will be removed in Flask-SQLAlchemy"728            " 3.2. Use 'engine' or 'engines[key]' instead. If you're using"729            " Flask-Migrate or Alembic, you'll need to update your 'env.py' file.",730            DeprecationWarning,731            stacklevel=2,732        )733 734        if "bind" in kwargs:735            bind_key = kwargs.pop("bind")736 737        return self.engines[bind_key]738 739    def get_or_404(740        self,741        entity: type[_O],742        ident: t.Any,743        *,744        description: str | None = None,745        **kwargs: t.Any,746    ) -> _O:747        """Like :meth:`session.get() <sqlalchemy.orm.Session.get>` but aborts with a748        ``404 Not Found`` error instead of returning ``None``.749 750        :param entity: The model class to query.751        :param ident: The primary key to query.752        :param description: A custom message to show on the error page.753        :param kwargs: Extra arguments passed to ``session.get()``.754 755        .. versionchanged:: 3.1756            Pass extra keyword arguments to ``session.get()``.757 758        .. versionadded:: 3.0759        """760        value = self.session.get(entity, ident, **kwargs)761 762        if value is None:763            abort(404, description=description)764 765        return value766 767    def first_or_404(768        self, statement: sa.sql.Select[t.Any], *, description: str | None = None769    ) -> t.Any:770        """Like :meth:`Result.scalar() <sqlalchemy.engine.Result.scalar>`, but aborts771        with a ``404 Not Found`` error instead of returning ``None``.772 773        :param statement: The ``select`` statement to execute.774        :param description: A custom message to show on the error page.775 776        .. versionadded:: 3.0777        """778        value = self.session.execute(statement).scalar()779 780        if value is None:781            abort(404, description=description)782 783        return value784 785    def one_or_404(786        self, statement: sa.sql.Select[t.Any], *, description: str | None = None787    ) -> t.Any:788        """Like :meth:`Result.scalar_one() <sqlalchemy.engine.Result.scalar_one>`,789        but aborts with a ``404 Not Found`` error instead of raising ``NoResultFound``790        or ``MultipleResultsFound``.791 792        :param statement: The ``select`` statement to execute.793        :param description: A custom message to show on the error page.794 795        .. versionadded:: 3.0796        """797        try:798            return self.session.execute(statement).scalar_one()799        except (sa_exc.NoResultFound, sa_exc.MultipleResultsFound):800            abort(404, description=description)801 802    def paginate(803        self,804        select: sa.sql.Select[t.Any],805        *,806        page: int | None = None,807        per_page: int | None = None,808        max_per_page: int | None = None,809        error_out: bool = True,810        count: bool = True,811    ) -> Pagination:812        """Apply an offset and limit to a select statment based on the current page and813        number of items per page, returning a :class:`.Pagination` object.814 815        The statement should select a model class, like ``select(User)``. This applies816        ``unique()`` and ``scalars()`` modifiers to the result, so compound selects will817        not return the expected results.818 819        :param select: The ``select`` statement to paginate.820        :param page: The current page, used to calculate the offset. Defaults to the821            ``page`` query arg during a request, or 1 otherwise.822        :param per_page: The maximum number of items on a page, used to calculate the823            offset and limit. Defaults to the ``per_page`` query arg during a request,824            or 20 otherwise.825        :param max_per_page: The maximum allowed value for ``per_page``, to limit a826            user-provided value. Use ``None`` for no limit. Defaults to 100.827        :param error_out: Abort with a ``404 Not Found`` error if no items are returned828            and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if829            either are not ints.830        :param count: Calculate the total number of values by issuing an extra count831            query. For very complex queries this may be inaccurate or slow, so it can be832            disabled and set manually if necessary.833 834        .. versionchanged:: 3.0835            The ``count`` query is more efficient.836 837        .. versionadded:: 3.0838        """839        return SelectPagination(840            select=select,841            session=self.session(),842            page=page,843            per_page=per_page,844            max_per_page=max_per_page,845            error_out=error_out,846            count=count,847        )848 849    def _call_for_binds(850        self, bind_key: str | None | list[str | None], op_name: str851    ) -> None:852        """Call a method on each metadata.853 854        :meta private:855 856        :param bind_key: A bind key or list of keys. Defaults to all binds.857        :param op_name: The name of the method to call.858 859        .. versionchanged:: 3.0860            Renamed from ``_execute_for_all_tables``.861        """862        if bind_key == "__all__":863            keys: list[str | None] = list(self.metadatas)864        elif bind_key is None or isinstance(bind_key, str):865            keys = [bind_key]866        else:867            keys = bind_key868 869        for key in keys:870            try:871                engine = self.engines[key]872            except KeyError:873                message = f"Bind key '{key}' is not in 'SQLALCHEMY_BINDS' config."874 875                if key is None:876                    message = f"'SQLALCHEMY_DATABASE_URI' config is not set. {message}"877 878                raise sa_exc.UnboundExecutionError(message) from None879 880            metadata = self.metadatas[key]881            getattr(metadata, op_name)(bind=engine)882 883    def create_all(self, bind_key: str | None | list[str | None] = "__all__") -> None:884        """Create tables that do not exist in the database by calling885        ``metadata.create_all()`` for all or some bind keys. This does not886        update existing tables, use a migration library for that.887 888        This requires that a Flask application context is active.889 890        :param bind_key: A bind key or list of keys to create the tables for. Defaults891            to all binds.892 893        .. versionchanged:: 3.0894            Renamed the ``bind`` parameter to ``bind_key``. Removed the ``app``895            parameter.896 897        .. versionchanged:: 0.12898            Added the ``bind`` and ``app`` parameters.899        """900        self._call_for_binds(bind_key, "create_all")901 902    def drop_all(self, bind_key: str | None | list[str | None] = "__all__") -> None:903        """Drop tables by calling ``metadata.drop_all()`` for all or some bind keys.904 905        This requires that a Flask application context is active.906 907        :param bind_key: A bind key or list of keys to drop the tables from. Defaults to908            all binds.909 910        .. versionchanged:: 3.0911            Renamed the ``bind`` parameter to ``bind_key``. Removed the ``app``912            parameter.913 914        .. versionchanged:: 0.12915            Added the ``bind`` and ``app`` parameters.916        """917        self._call_for_binds(bind_key, "drop_all")918 919    def reflect(self, bind_key: str | None | list[str | None] = "__all__") -> None:920        """Load table definitions from the database by calling ``metadata.reflect()``921        for all or some bind keys.922 923        This requires that a Flask application context is active.924 925        :param bind_key: A bind key or list of keys to reflect the tables from. Defaults926            to all binds.927 928        .. versionchanged:: 3.0929            Renamed the ``bind`` parameter to ``bind_key``. Removed the ``app``930            parameter.931 932        .. versionchanged:: 0.12933            Added the ``bind`` and ``app`` parameters.934        """935        self._call_for_binds(bind_key, "reflect")936 937    def _set_rel_query(self, kwargs: dict[str, t.Any]) -> None:938        """Apply the extension's :attr:`Query` class as the default for relationships939        and backrefs.940 941        :meta private:942        """943        kwargs.setdefault("query_class", self.Query)944 945        if "backref" in kwargs:946            backref = kwargs["backref"]947 948            if isinstance(backref, str):949                backref = (backref, {})950 951            backref[1].setdefault("query_class", self.Query)952 953    def relationship(954        self, *args: t.Any, **kwargs: t.Any955    ) -> sa_orm.RelationshipProperty[t.Any]:956        """A :func:`sqlalchemy.orm.relationship` that applies this extension's957        :attr:`Query` class for dynamic relationships and backrefs.958 959        .. versionchanged:: 3.0960            The :attr:`Query` class is set on ``backref``.961        """962        self._set_rel_query(kwargs)963        return sa_orm.relationship(*args, **kwargs)964 965    def dynamic_loader(966        self, argument: t.Any, **kwargs: t.Any967    ) -> sa_orm.RelationshipProperty[t.Any]:968        """A :func:`sqlalchemy.orm.dynamic_loader` that applies this extension's969        :attr:`Query` class for relationships and backrefs.970 971        .. versionchanged:: 3.0972            The :attr:`Query` class is set on ``backref``.973        """974        self._set_rel_query(kwargs)975        return sa_orm.dynamic_loader(argument, **kwargs)976 977    def _relation(978        self, *args: t.Any, **kwargs: t.Any979    ) -> sa_orm.RelationshipProperty[t.Any]:980        """A :func:`sqlalchemy.orm.relationship` that applies this extension's981        :attr:`Query` class for dynamic relationships and backrefs.982 983        SQLAlchemy 2.0 removes this name, use ``relationship`` instead.984 985        :meta private:986 987        .. versionchanged:: 3.0988            The :attr:`Query` class is set on ``backref``.989        """990        self._set_rel_query(kwargs)991        f = sa_orm.relationship992        return f(*args, **kwargs)993 994    def __getattr__(self, name: str) -> t.Any:995        if name == "relation":996            return self._relation997 998        if name == "event":999            return sa_event1000 1001        if name.startswith("_"):1002            raise AttributeError(name)1003 1004        for mod in (sa, sa_orm):1005            if hasattr(mod, name):1006                return getattr(mod, name)1007 1008        raise AttributeError(name)1009