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

93.83

/src/coaster/sqlalchemy/comparators.py

"""Enhanced query and custom comparators."""

from __future__ import annotations

import contextlib
from collections.abc import Iterator
from typing import Any, Optional, Union
from uuid import UUID

import sqlalchemy as sa
from sqlalchemy.ext.hybrid import Comparator

from ..utils import uuid_from_base58, uuid_from_base64

__all__ = [
    'SplitIndexComparator',
    'SqlSplitIdComparator',
    'SqlUuidHexComparator',
    'SqlUuidB64Comparator',
    'SqlUuidB58Comparator',
]


class SplitIndexComparator(Comparator):
    """Base class for comparators that split a string and compare with one part."""

    def __init__(
        self,
        expression: Any,
        splitindex: Optional[int] = None,
        separator: str = '-',
    ) -> None:
        super().__init__(expression)
        self.splitindex = splitindex
        self.separator = separator

    def _decode(self, other: str) -> Any:
        raise NotImplementedError

    def __eq__(self, other: object) -> sa.ColumnElement[bool]:  # type: ignore[override]
        try:
            other = self._decode(other)  # type: ignore[arg-type]
        except (ValueError, TypeError):
            # If other could not be decoded, we do not match.
            return sa.sql.expression.false()
        return self.__clause_element__() == other  # type: ignore[return-value]

    is_ = __eq__  # type: ignore[assignment]

    def __ne__(self, other: object) -> sa.ColumnElement[bool]:  # type: ignore[override]
        try:
            other = self._decode(other)  # type: ignore[arg-type]
        except (ValueError, TypeError):
            # If other could not be decoded, we are not equal.
            return sa.sql.expression.true()
        return self.__clause_element__() != other  # type: ignore[return-value]

    isnot = __ne__  # type: ignore[assignment]
    is_not = __ne__  # type: ignore[assignment]

    def in_(self, other: Any) -> sa.ColumnElement[bool]:  # type: ignore[override]
        """Check if self is present in the other."""

        def errordecode(otherlist: Any) -> Iterator[str]:
            for val in otherlist:
                with contextlib.suppress(ValueError, TypeError):
                    yield self._decode(val)

        valid_values = list(errordecode(other))
        if not valid_values:
            # If none of the elements could be decoded, return false
            return sa.sql.expression.false()

        return self.__clause_element__().in_(valid_values)  # type: ignore[attr-defined]


class SqlSplitIdComparator(SplitIndexComparator):
    """
    Given an ``id-text`` string, split out the integer id and allows comparison on it.

    Also supports ``text-id``, ``text-text-id`` or other specific locations for the id
    if specified as a `splitindex` parameter to the constructor.

    This comparator will not attempt to decode non-string values, and will attempt to
    support all operators, accepting SQL expressions for :attr:`other`.
    """

    def _decode(self, other: Any) -> Union[int, Any]:
        if isinstance(other, str):
            if self.splitindex is not None:
                return int(other.split(self.separator)[self.splitindex])
            return int(other)
        return other

    # FIXME: The type of `op` is not known as the sample code is not type-annotated in
    # https://docs.sqlalchemy.org/en/20/orm/extensions/hybrid.html
    # #building-custom-comparators
    def operate(self, op: Any, *other: Any, **kwargs) -> sa.ColumnElement[Any]:
        """Perform SQL operation on decoded value for other."""
        # If `other` cannot be decoded, this operation will raise a Python exception
        return op(
            self.__clause_element__(), *(self._decode(o) for o in other), **kwargs
        )


class SqlUuidHexComparator(SplitIndexComparator):
    """
    Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.

    The UUID must be in hex format.

    Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
    UUID value if specified as a `splitindex` parameter to the constructor.
    """

    def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:
        if other is None:
            return None
        if not isinstance(other, UUID):
            if self.splitindex is not None:
                other = other.split(self.separator)[self.splitindex]
            return UUID(other)
        return other


class SqlUuidB64Comparator(SplitIndexComparator):
    """
    Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.

    The UUID must be in URL-safe Base64 format.

    Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
    UUID value if specified as a `splitindex` parameter to the constructor.

    Note that the default separator from the base class is ``-``, which is also a
    valid character in URL-safe Base64, so a custom separator must be specified when
    using this comparator.
    """

    def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:
        if other is None:
            return None
        if not isinstance(other, UUID):
            if self.splitindex is not None:
                other = other.split(self.separator)[self.splitindex]
            return uuid_from_base64(other)
        return other


class SqlUuidB58Comparator(SplitIndexComparator):
    """
    Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.

    The UUID must be in Base58 format.

    Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
    UUID value if specified as a `splitindex` parameter to the constructor.
    """

    def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:
        if other is None:
            return None
        if not isinstance(other, UUID):
            if self.splitindex is not None:
                other = other.split('-')[self.splitindex]
            return uuid_from_base58(other)
        return other

1	"""Enhanced query and custom comparators."""	4✔
2
3	from __future__ import annotations	4✔
4
5	import contextlib	4✔
6	from collections.abc import Iterator	4✔
7	from typing import Any, Optional, Union	4✔
8	from uuid import UUID	4✔
9
10	import sqlalchemy as sa	4✔
11	from sqlalchemy.ext.hybrid import Comparator	4✔
12
13	from ..utils import uuid_from_base58, uuid_from_base64	4✔
14
15	__all__ = [	4✔
16	'SplitIndexComparator',
17	'SqlSplitIdComparator',
18	'SqlUuidHexComparator',
19	'SqlUuidB64Comparator',
20	'SqlUuidB58Comparator',
21	]
22
23
24	class SplitIndexComparator(Comparator):	4✔
25	"""Base class for comparators that split a string and compare with one part."""	4✔
26
27	def __init__(	4✔
28	self,
29	expression: Any,
30	splitindex: Optional[int] = None,
31	separator: str = '-',
32	) -> None:
33	super().__init__(expression)	4✔
34	self.splitindex = splitindex	4✔
35	self.separator = separator	4✔
36
37	def _decode(self, other: str) -> Any:	4✔
38	raise NotImplementedError
39
40	def __eq__(self, other: object) -> sa.ColumnElement[bool]: # type: ignore[override]	4✔
41	try:	4✔
42	other = self._decode(other) # type: ignore[arg-type]	4✔
43	except (ValueError, TypeError):	4✔
44	# If other could not be decoded, we do not match.
45	return sa.sql.expression.false()	4✔
46	return self.__clause_element__() == other # type: ignore[return-value]	4✔
47
48	is_ = __eq__ # type: ignore[assignment]	4✔
49
50	def __ne__(self, other: object) -> sa.ColumnElement[bool]: # type: ignore[override]	4✔
51	try:	4✔
52	other = self._decode(other) # type: ignore[arg-type]	4✔
53	except (ValueError, TypeError):	4✔
54	# If other could not be decoded, we are not equal.
55	return sa.sql.expression.true()	4✔
56	return self.__clause_element__() != other # type: ignore[return-value]	4✔
57
58	isnot = __ne__ # type: ignore[assignment]	4✔
59	is_not = __ne__ # type: ignore[assignment]	4✔
60
61	def in_(self, other: Any) -> sa.ColumnElement[bool]: # type: ignore[override]	4✔
62	"""Check if self is present in the other."""
63
64	def errordecode(otherlist: Any) -> Iterator[str]:	4✔
65	for val in otherlist:	4✔
66	with contextlib.suppress(ValueError, TypeError):	4✔
67	yield self._decode(val)	4✔
68
69	valid_values = list(errordecode(other))	4✔
70	if not valid_values:	4✔
71	# If none of the elements could be decoded, return false
UNCOV 72	return sa.sql.expression.false()	×
73
74	return self.__clause_element__().in_(valid_values) # type: ignore[attr-defined]	4✔
75
76
77	class SqlSplitIdComparator(SplitIndexComparator):	4✔
78	"""	4✔
79	Given an ``id-text`` string, split out the integer id and allows comparison on it.
80
81	Also supports ``text-id``, ``text-text-id`` or other specific locations for the id
82	if specified as a `splitindex` parameter to the constructor.
83
84	This comparator will not attempt to decode non-string values, and will attempt to
85	support all operators, accepting SQL expressions for :attr:`other`.
86	"""
87
88	def _decode(self, other: Any) -> Union[int, Any]:	4✔
89	if isinstance(other, str):	4✔
90	if self.splitindex is not None:	4✔
91	return int(other.split(self.separator)[self.splitindex])	4✔
92	return int(other)	4✔
93	return other	4✔
94
95	# FIXME: The type of `op` is not known as the sample code is not type-annotated in
96	# https://docs.sqlalchemy.org/en/20/orm/extensions/hybrid.html
97	# #building-custom-comparators
98	def operate(self, op: Any, other: Any, *kwargs) -> sa.ColumnElement[Any]:	4✔
99	"""Perform SQL operation on decoded value for other."""
100	# If `other` cannot be decoded, this operation will raise a Python exception
UNCOV 101	return op(	×
102	self.__clause_element__(), (self._decode(o) for o in other), *kwargs
103	)
104
105
106	class SqlUuidHexComparator(SplitIndexComparator):	4✔
107	"""	4✔
108	Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.
109
110	The UUID must be in hex format.
111
112	Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
113	UUID value if specified as a `splitindex` parameter to the constructor.
114	"""
115
116	def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:	4✔
117	if other is None:	4✔
118	return None	4✔
119	if not isinstance(other, UUID):	4✔
120	if self.splitindex is not None:	4✔
121	other = other.split(self.separator)[self.splitindex]	4✔
122	return UUID(other)	4✔
123	return other	4✔
124
125
126	class SqlUuidB64Comparator(SplitIndexComparator):	4✔
127	"""	4✔
128	Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.
129
130	The UUID must be in URL-safe Base64 format.
131
132	Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
133	UUID value if specified as a `splitindex` parameter to the constructor.
134
135	Note that the default separator from the base class is ``-``, which is also a
136	valid character in URL-safe Base64, so a custom separator must be specified when
137	using this comparator.
138	"""
139
140	def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:	4✔
141	if other is None:	4✔
142	return None	4✔
143	if not isinstance(other, UUID):	4✔
144	if self.splitindex is not None:	4✔
UNCOV 145	other = other.split(self.separator)[self.splitindex]	×
146	return uuid_from_base64(other)	4✔
UNCOV 147	return other	×
148
149
150	class SqlUuidB58Comparator(SplitIndexComparator):	4✔
151	"""	4✔
152	Given an ``uuid-text`` string, split out the UUID and allow comparisons on it.
153
154	The UUID must be in Base58 format.
155
156	Also supports ``text-uuid``, ``text-text-uuid`` or other specific locations for the
157	UUID value if specified as a `splitindex` parameter to the constructor.
158	"""
159
160	def _decode(self, other: Optional[Union[str, UUID]]) -> Optional[UUID]:	4✔
161	if other is None:	4✔
162	return None	4✔
163	if not isinstance(other, UUID):	4✔
164	if self.splitindex is not None:	4✔
165	other = other.split('-')[self.splitindex]	4✔
166	return uuid_from_base58(other)	4✔
UNCOV 167	return other	×

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