9244418196

Committed 26 May 2024 03:39PM UTC coverage: 84.467% (-4.8%) from 89.263%

Build # 9244418196

Build Type

push

github

Committed by

web-flow

Commit Message

Add async support for Quart+Flask (#470)

This commit bumps the version number from 0.7 to 0.8 as it has extensive changes:

* Ruff replaces black, isort and flake8 for linting and formatting
* All decorators now support async functions and provide async wrapper implementations
* Some obsolete modules have been removed
* Pagination from Flask-SQLAlchemy is now included, removing that dependency (but still used in tests)
* New `compat` module provides wrappers to both Quart and Flake and is used by all other modules
* Some tests run using Quart. The vast majority of tests are not upgraded, nor are there tests for async decorators, so overall line coverage has dropped significantly. Comprehensive test coverage is still pending; for now we are using Funnel's tests as the extended test suite

Run Details

648 of 1023 new or added lines in 29 files covered. (63.34%)

138 existing lines in 17 files now uncovered.

3948 of 4674 relevant lines covered (84.47%)

3.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.14

/src/coaster/sqlalchemy/query.py

"""Legacy Query API with additional methods."""

from __future__ import annotations

import warnings
from collections.abc import Collection
from functools import wraps
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    Optional,
    TypeVar,
    Union,
    cast,
    overload,
)
from typing_extensions import ParamSpec

from sqlalchemy import ColumnExpressionArgument
from sqlalchemy.exc import NoResultFound
from sqlalchemy.orm import (
    DynamicMapped as DynamicMappedBase,
    InstrumentedAttribute,
    Query as QueryBase,
    Relationship as RelationshipBase,
    backref as backref_base,
    relationship as relationship_base,
)
from sqlalchemy.orm.dynamic import AppenderMixin

from ..compat import abort
from .pagination import QueryPagination

__all__ = [
    'BackrefWarning',
    'ModelWarning',
    'Query',
    'AppenderQuery',
    'QueryProperty',
    'DynamicMapped',
    'Relationship',
    'relationship',
    'backref',
]

_T = TypeVar('_T', bound=Any)
_T_co = TypeVar("_T_co", bound=Any, covariant=True)

# --- Warnings -------------------------------------------------------------------------


class BackrefWarning(UserWarning):
    """Warning for type-unfriendly use of ``backref`` in a :func:`relationship`."""


# Legacy name, do not use in new code
ModelWarning = BackrefWarning


# --- Query class and property ---------------------------------------------------------
# Change Query's Generic type to be covariant. This needed because:
# 1. When using SQLAlchemy polymorphism, a query on the base type may return a subtype.
# 2. For typing, a classmethod that returns Query[Self] will be deemed incompatible with
#    Query[HostModel] as Query[HostModel] != Query[Self@HostModel] because Self could be
#    a subclass.
class Query(QueryBase[_T_co]):  # type: ignore[type-var]
    """Extends SQLAlchemy's :class:`~sqlalchemy.orm.Query` with additional methods."""

    if TYPE_CHECKING:
        # The calls to super() here will never happen. They are to aid the programmer
        # using an editor's "Go to Definition" feature

        def get(self, ident: Any) -> Optional[_T_co]:
            """Provide type hint certifying that `get` returns `_T_co | None`."""
            return super().get(ident)

        def first(self) -> Optional[_T_co]:
            """Provide type hint certifying that `first` returns `_T_co | None`."""
            return super().first()

        def one(self) -> _T_co:
            """Provide type hint certifying that `one` returns `_T_co`."""
            return super().one()

        def add_columns(self, *column: ColumnExpressionArgument[Any]) -> Query[Any]:
            """Fix type hint to refer to :class:`Query`."""
            # pylint: disable=useless-parent-delegation
            return super().add_columns(*column)  # type: ignore[return-value]

        def with_transformation(
            self, fn: Callable[[QueryBase[Any]], QueryBase[Any]]
        ) -> Query[Any]:
            """Fix type hint to refer to :class:`Query`."""
            return super().with_transformation(fn)  # type: ignore[return-value]

    def get_or_404(self, ident: Any, description: Optional[str] = None) -> _T_co:
        """
        Like :meth:`~sqlalchemy.orm.Query.get` but aborts with 404 if no result.

        :param ident: The primary key to query
        :param description: A custom message to show on the error page
        """
        rv = self.get(ident)

        if rv is None:
            abort(404, description=description)

        return rv

    def first_or_404(self, description: Optional[str] = None) -> _T_co:
        """
        Like :meth:`~sqlalchemy.orm.Query.first` but aborts with 404 if no result.

        :param description: A custom message to show on the error page
        """
        rv = self.first()

        if rv is None:
            abort(404, description=description)

        return rv

    def one_or_404(self, description: Optional[str] = None) -> _T_co:
        """
        Like :meth:`~sqlalchemy.orm.Query.one`, but aborts with 404 for NoResultFound.

        Unlike Flask-SQLAlchemy's implementation,
        :exc:`~sqlalchemy.exc.MultipleResultsFound` is not recast as 404 and will cause
        a 500 error if not handled. The query may need additional filters to target a
        single result.

        :param description: A custom message to show on the error page
        """
        try:
            return self.one()
        except NoResultFound:
            abort(404, description=description)
        # Pylint doesn't know abort is NoReturn
        return None  # type: ignore[unreachable]

    def notempty(self) -> bool:
        """
        Return `True` if the query has non-zero results.

        Does the equivalent of ``bool(query.count())`` but using an efficient
        SQL EXISTS operator, so the database stops counting after the first result
        is found.
        """
        return self.session.query(self.exists()).scalar()

    def isempty(self) -> bool:
        """
        Return `True` if the query has zero results.

        Does the equivalent of ``not bool(query.count())`` but using an efficient
        SQL EXISTS operator, so the database stops counting after the first result
        is found.
        """
        return not self.session.query(self.exists()).scalar()

    # TODO: Pagination may not preserve model type information, affecting downstream
    # type validation
    def paginate(
        self,
        *,
        page: int | None = None,
        per_page: int | None = None,
        max_per_page: int | None = None,
        error_out: bool = True,
        count: bool = True,
    ) -> QueryPagination[_T_co]:
        """
        Apply an offset and limit to the query, returning a Pagination object.

        :param page: The current page, used to calculate the offset. Defaults to the
            ``page`` query arg during a request, or 1 otherwise
        :param per_page: The maximum number of items on a page, used to calculate the
            offset and limit. Defaults to the ``per_page`` query arg during a request,
            or 20 otherwise
        :param max_per_page: The maximum allowed value for ``per_page``, to limit a
            user-provided value. Use ``None`` for no limit. Defaults to 100
        :param error_out: Abort with a ``404 Not Found`` error if no items are returned
            and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if
            either are not ints
        :param count: Calculate the total number of values by issuing an extra count
            query. For very complex queries this may be inaccurate or slow, so it can be
            disabled and set manually if necessary
        """
        return QueryPagination(
            query=self,
            page=page,
            per_page=per_page,
            max_per_page=max_per_page,
            error_out=error_out,
            count=count,
        )


# AppenderMixin and Query have different definitions for ``session``, so we have to ask
# Mypy to ignore it [misc]. SQLAlchemy defines the generic type as invariant but we
# change it to covariant, so we need an ignore that too [type-var].
class AppenderQuery(AppenderMixin[_T_co], Query[_T_co]):  # type: ignore[misc,type-var]
    """
    AppenderQuery, used by :func:`relationship` as the default query class.

    SQLAlchemy's :func:`~sqlalchemy.orm.relationship` will accept ``query_class=Query``
    directly, but will construct a new class mixing
    :func:`~sqlalchemy.orm.dynamic.AppenderMixin` and ``query_class`` if
    ``AppenderMixin`` is not in the existing base classes.
    """

    # AppenderMixin does not specify a type for query_class
    query_class: Optional[type[Query[_T_co]]] = Query


class QueryProperty:
    """A class property that creates a query object for a model."""

    def __get__(self, _obj: Optional[_T_co], cls: type[_T_co]) -> Query[_T_co]:
        return cls.query_class(cls, session=cls.__fsa__.session())


# --- `relationship` and `backref` wrappers for `lazy='dynamic'` -----------------------

# DynamicMapped and Relationship are redefined from the original in SQLAlchemy to offer
# a type hint to Coaster's AppenderQuery, which in turn wraps Coaster's Query with its
# additional methods

if TYPE_CHECKING:

    class DynamicMapped(DynamicMappedBase[_T_co]):
        """Represent the ORM mapped attribute type for a "dynamic" relationship."""

        __slots__ = ()

        @overload  # type: ignore[override]
        def __get__(
            self, instance: None, owner: Any
        ) -> InstrumentedAttribute[_T_co]: ...

        @overload
        def __get__(self, instance: object, owner: Any) -> AppenderQuery[_T_co]: ...

        def __get__(
            self, instance: Optional[object], owner: Any
        ) -> Union[InstrumentedAttribute[_T_co], AppenderQuery[_T_co]]: ...

        def __set__(self, instance: Any, value: Collection[_T_co]) -> None: ...

    class Relationship(RelationshipBase[_T], DynamicMapped[_T]):  # type: ignore[misc]
        """Wraps Relationship with the updated version of DynamicMapped."""

else:
    # Avoid the overhead of empty subclasses at runtime
    DynamicMapped = DynamicMappedBase
    Relationship = RelationshipBase


_P = ParamSpec('_P')


# This wrapper exists solely for type hinting tools as @wraps itself does not
# provide type hints indicating that the function's type signature is unchanged
def _create_relationship_wrapper(f: Callable[_P, Any]) -> Callable[_P, Relationship]:
    """Create a wrapper for relationship."""

    @wraps(f)
    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> Relationship:
        """Insert a default query_class when constructing a relationship."""
        if 'query_class' not in kwargs:
            kwargs['query_class'] = AppenderQuery
        if 'backref' in kwargs:
            warnings.warn(
                "`backref` is not compatible with type hinting. Use `back_populates`:"
                " https://docs.sqlalchemy.org/en/20/orm/backref.html",
                BackrefWarning,
                stacklevel=2,
            )
        return cast(Relationship, f(*args, **kwargs))

    return wrapper


# `backref` does not change return type, unlike `relationship`
def _create_backref_wrapper(f: Callable[_P, _T]) -> Callable[_P, _T]:
    """Create a wrapper for `backref`."""

    @wraps(f)
    def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _T:
        """Insert a default query_class when constructing a `backref`."""
        if 'query_class' not in kwargs:
            kwargs['query_class'] = AppenderQuery
        return f(*args, **kwargs)

    return wrapper


#: Wrap :func:`~sqlalchemy.orm.relationship` to insert :class:`Query` as the default
#: value for :attr:`query_class`
relationship = _create_relationship_wrapper(relationship_base)
#: Wrap :func:`~sqlalchemy.orm.backref` to insert :class:`Query` as the default
#: value for :attr:`query_class`
backref = _create_backref_wrapper(backref_base)

1	"""Legacy Query API with additional methods."""	4✔
2
3	from __future__ import annotations	4✔
4
5	import warnings	4✔
6	from collections.abc import Collection	4✔
7	from functools import wraps	4✔
8	from typing import (	4✔
9	TYPE_CHECKING,
10	Any,
11	Callable,
12	Optional,
13	TypeVar,
14	Union,
15	cast,
16	overload,
17	)
18	from typing_extensions import ParamSpec	4✔
19
20	from sqlalchemy import ColumnExpressionArgument	4✔
21	from sqlalchemy.exc import NoResultFound	4✔
22	from sqlalchemy.orm import (	4✔
23	DynamicMapped as DynamicMappedBase,
24	InstrumentedAttribute,
25	Query as QueryBase,
26	Relationship as RelationshipBase,
27	backref as backref_base,
28	relationship as relationship_base,
29	)
30	from sqlalchemy.orm.dynamic import AppenderMixin	4✔
31
32	from ..compat import abort	4✔
33	from .pagination import QueryPagination	4✔
34
35	__all__ = [	4✔
36	'BackrefWarning',
37	'ModelWarning',
38	'Query',
39	'AppenderQuery',
40	'QueryProperty',
41	'DynamicMapped',
42	'Relationship',
43	'relationship',
44	'backref',
45	]
46
47	_T = TypeVar('_T', bound=Any)	4✔
48	_T_co = TypeVar("_T_co", bound=Any, covariant=True)	4✔
49
50	# --- Warnings -------------------------------------------------------------------------
51
52
53	class BackrefWarning(UserWarning):	4✔
54	"""Warning for type-unfriendly use of ``backref`` in a :func:`relationship`."""	4✔
55
56
57	# Legacy name, do not use in new code
58	ModelWarning = BackrefWarning	4✔
59
60
61	# --- Query class and property ---------------------------------------------------------
62	# Change Query's Generic type to be covariant. This needed because:
63	# 1. When using SQLAlchemy polymorphism, a query on the base type may return a subtype.
64	# 2. For typing, a classmethod that returns Query[Self] will be deemed incompatible with
65	# Query[HostModel] as Query[HostModel] != Query[Self@HostModel] because Self could be
66	# a subclass.
67	class Query(QueryBase[_T_co]): # type: ignore[type-var]	4✔
68	"""Extends SQLAlchemy's :class:`~sqlalchemy.orm.Query` with additional methods."""	4✔
69
70	if TYPE_CHECKING:
71	# The calls to super() here will never happen. They are to aid the programmer
72	# using an editor's "Go to Definition" feature
73
74	def get(self, ident: Any) -> Optional[_T_co]:
75	"""Provide type hint certifying that `get` returns `_T_co \| None`."""
76	return super().get(ident)
77
78	def first(self) -> Optional[_T_co]:
79	"""Provide type hint certifying that `first` returns `_T_co \| None`."""
80	return super().first()
81
82	def one(self) -> _T_co:
83	"""Provide type hint certifying that `one` returns `_T_co`."""
84	return super().one()
85
86	def add_columns(self, *column: ColumnExpressionArgument[Any]) -> Query[Any]:
87	"""Fix type hint to refer to :class:`Query`."""
88	# pylint: disable=useless-parent-delegation
89	return super().add_columns(*column) # type: ignore[return-value]
90
91	def with_transformation(
92	self, fn: Callable[[QueryBase[Any]], QueryBase[Any]]
93	) -> Query[Any]:
94	"""Fix type hint to refer to :class:`Query`."""
95	return super().with_transformation(fn) # type: ignore[return-value]
96
97	def get_or_404(self, ident: Any, description: Optional[str] = None) -> _T_co:	4✔
98	"""
99	Like :meth:`~sqlalchemy.orm.Query.get` but aborts with 404 if no result.
100
101	:param ident: The primary key to query
102	:param description: A custom message to show on the error page
103	"""
NEW 104	rv = self.get(ident)	×
105
NEW 106	if rv is None:	×
NEW 107	abort(404, description=description)	×
108
NEW 109	return rv	×
110
111	def first_or_404(self, description: Optional[str] = None) -> _T_co:	4✔
112	"""
113	Like :meth:`~sqlalchemy.orm.Query.first` but aborts with 404 if no result.
114
115	:param description: A custom message to show on the error page
116	"""
117	rv = self.first()	4✔
118
119	if rv is None:	4✔
120	abort(404, description=description)	4✔
121
122	return rv	4✔
123
124	def one_or_404(self, description: Optional[str] = None) -> _T_co:	4✔
125	"""
126	Like :meth:`~sqlalchemy.orm.Query.one`, but aborts with 404 for NoResultFound.
127
128	Unlike Flask-SQLAlchemy's implementation,
129	:exc:`~sqlalchemy.exc.MultipleResultsFound` is not recast as 404 and will cause
130	a 500 error if not handled. The query may need additional filters to target a
131	single result.
132
133	:param description: A custom message to show on the error page
134	"""
135	try:	4✔
136	return self.one()	4✔
137	except NoResultFound:	4✔
138	abort(404, description=description)	4✔
139	# Pylint doesn't know abort is NoReturn
NEW 140	return None # type: ignore[unreachable]	×
141
142	def notempty(self) -> bool:	4✔
143	"""
144	Return `True` if the query has non-zero results.
145
146	Does the equivalent of ``bool(query.count())`` but using an efficient
147	SQL EXISTS operator, so the database stops counting after the first result
148	is found.
149	"""
150	return self.session.query(self.exists()).scalar()	4✔
151
152	def isempty(self) -> bool:	4✔
153	"""
154	Return `True` if the query has zero results.
155
156	Does the equivalent of ``not bool(query.count())`` but using an efficient
157	SQL EXISTS operator, so the database stops counting after the first result
158	is found.
159	"""
NEW 160	return not self.session.query(self.exists()).scalar()	×
161
162	# TODO: Pagination may not preserve model type information, affecting downstream
163	# type validation
164	def paginate(	4✔
165	self,
166	*,
167	page: int \| None = None,
168	per_page: int \| None = None,
169	max_per_page: int \| None = None,
170	error_out: bool = True,
171	count: bool = True,
172	) -> QueryPagination[_T_co]:
173	"""
174	Apply an offset and limit to the query, returning a Pagination object.
175
176	:param page: The current page, used to calculate the offset. Defaults to the
177	``page`` query arg during a request, or 1 otherwise
178	:param per_page: The maximum number of items on a page, used to calculate the
179	offset and limit. Defaults to the ``per_page`` query arg during a request,
180	or 20 otherwise
181	:param max_per_page: The maximum allowed value for ``per_page``, to limit a
182	user-provided value. Use ``None`` for no limit. Defaults to 100
183	:param error_out: Abort with a ``404 Not Found`` error if no items are returned
184	and ``page`` is not 1, or if ``page`` or ``per_page`` is less than 1, or if
185	either are not ints
186	:param count: Calculate the total number of values by issuing an extra count
187	query. For very complex queries this may be inaccurate or slow, so it can be
188	disabled and set manually if necessary
189	"""
NEW 190	return QueryPagination(	×
191	query=self,
192	page=page,
193	per_page=per_page,
194	max_per_page=max_per_page,
195	error_out=error_out,
196	count=count,
197	)
198
199
200	# AppenderMixin and Query have different definitions for ``session``, so we have to ask
201	# Mypy to ignore it [misc]. SQLAlchemy defines the generic type as invariant but we
202	# change it to covariant, so we need an ignore that too [type-var].
203	class AppenderQuery(AppenderMixin[_T_co], Query[_T_co]): # type: ignore[misc,type-var]	4✔
204	"""	4✔
205	AppenderQuery, used by :func:`relationship` as the default query class.
206
207	SQLAlchemy's :func:`~sqlalchemy.orm.relationship` will accept ``query_class=Query``
208	directly, but will construct a new class mixing
209	:func:`~sqlalchemy.orm.dynamic.AppenderMixin` and ``query_class`` if
210	``AppenderMixin`` is not in the existing base classes.
211	"""
212
213	# AppenderMixin does not specify a type for query_class
214	query_class: Optional[type[Query[_T_co]]] = Query	4✔
215
216
217	class QueryProperty:	4✔
218	"""A class property that creates a query object for a model."""	4✔
219
220	def __get__(self, _obj: Optional[_T_co], cls: type[_T_co]) -> Query[_T_co]:	4✔
221	return cls.query_class(cls, session=cls.__fsa__.session())	4✔
222
223
224	# --- `relationship` and `backref` wrappers for `lazy='dynamic'` -----------------------
225
226	# DynamicMapped and Relationship are redefined from the original in SQLAlchemy to offer
227	# a type hint to Coaster's AppenderQuery, which in turn wraps Coaster's Query with its
228	# additional methods
229
230	if TYPE_CHECKING:
231
232	class DynamicMapped(DynamicMappedBase[_T_co]):
233	"""Represent the ORM mapped attribute type for a "dynamic" relationship."""
234
235	__slots__ = ()
236
237	@overload # type: ignore[override]
238	def __get__(
239	self, instance: None, owner: Any
240	) -> InstrumentedAttribute[_T_co]: ...
241
242	@overload
243	def __get__(self, instance: object, owner: Any) -> AppenderQuery[_T_co]: ...
244
245	def __get__(
246	self, instance: Optional[object], owner: Any
247	) -> Union[InstrumentedAttribute[_T_co], AppenderQuery[_T_co]]: ...
248
249	def __set__(self, instance: Any, value: Collection[_T_co]) -> None: ...
250
251	class Relationship(RelationshipBase[_T], DynamicMapped[_T]): # type: ignore[misc]
252	"""Wraps Relationship with the updated version of DynamicMapped."""
253
254	else:
255	# Avoid the overhead of empty subclasses at runtime
256	DynamicMapped = DynamicMappedBase	4✔
257	Relationship = RelationshipBase	4✔
258
259
260	_P = ParamSpec('_P')	4✔
261
262
263	# This wrapper exists solely for type hinting tools as @wraps itself does not
264	# provide type hints indicating that the function's type signature is unchanged
265	def _create_relationship_wrapper(f: Callable[_P, Any]) -> Callable[_P, Relationship]:	4✔
266	"""Create a wrapper for relationship."""
267
268	@wraps(f)	4✔
269	def wrapper(args: _P.args, *kwargs: _P.kwargs) -> Relationship:	4✔
270	"""Insert a default query_class when constructing a relationship."""
271	if 'query_class' not in kwargs:	4✔
272	kwargs['query_class'] = AppenderQuery	4✔
273	if 'backref' in kwargs:	4✔
274	warnings.warn(	4✔
275	"`backref` is not compatible with type hinting. Use `back_populates`:"
276	" https://docs.sqlalchemy.org/en/20/orm/backref.html",
277	BackrefWarning,
278	stacklevel=2,
279	)
280	return cast(Relationship, f(args, *kwargs))	4✔
281
282	return wrapper	4✔
283
284
285	# `backref` does not change return type, unlike `relationship`
286	def _create_backref_wrapper(f: Callable[_P, _T]) -> Callable[_P, _T]:	4✔
287	"""Create a wrapper for `backref`."""
288
289	@wraps(f)	4✔
290	def wrapper(args: _P.args, *kwargs: _P.kwargs) -> _T:	4✔
291	"""Insert a default query_class when constructing a `backref`."""
292	if 'query_class' not in kwargs:	4✔
293	kwargs['query_class'] = AppenderQuery	4✔
294	return f(args, *kwargs)	4✔
295
296	return wrapper	4✔
297
298
299	#: Wrap :func:`~sqlalchemy.orm.relationship` to insert :class:`Query` as the default
300	#: value for :attr:`query_class`
301	relationship = _create_relationship_wrapper(relationship_base)	4✔
302	#: Wrap :func:`~sqlalchemy.orm.backref` to insert :class:`Query` as the default
303	#: value for :attr:`query_class`
304	backref = _create_backref_wrapper(backref_base)	4✔

hasgeek / coaster / 9244418196

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous