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

92.68

/src/coaster/sqlalchemy/columns.py

"""SQLAlchemy column types."""

from __future__ import annotations

import json
from collections.abc import Collection, Mapping
from typing import Any, Optional

import sqlalchemy as sa
from furl import furl
from sqlalchemy.dialects import postgresql
from sqlalchemy.ext.mutable import Mutable
from sqlalchemy.types import TypeDecorator
from sqlalchemy_utils.types import URLType as UrlTypeBase

__all__ = ['JsonDict', 'UrlType']


# Adapted from http://docs.sqlalchemy.org/en/rel_0_8/orm/extensions/mutable.html
# #establishing-mutability-on-scalar-column-values


class JsonDict(TypeDecorator):
    """
    Represents a JSON data structure.

    Usage::

        column = Column(JsonDict)

    The column will be represented to the database as a ``JSONB`` column if
    the server is PostgreSQL 9.4 or later, ``JSON`` if PostgreSQL 9.2 or 9.3,
    and ``TEXT`` for everything else. The column behaves like a JSON store
    regardless of the backing data type.
    """

    # TypeDecorator replaces the Type with an instance of the type in the instance
    impl: sa.types.JSON = sa.types.JSON  # type: ignore[assignment]
    cache_ok = False

    def load_dialect_impl(self, dialect: sa.Dialect) -> sa.types.TypeEngine:
        """Use JSONB column in PostgreSQL."""
        if dialect.name == 'postgresql':
            return dialect.type_descriptor(postgresql.JSONB)  # type: ignore[arg-type]
        return dialect.type_descriptor(self.impl)

    def coerce_compared_value(self, op: Any, value: Any) -> sa.types.TypeEngine:
        """Coerce an incoming value using the JSON type's default handler."""
        return self.impl.coerce_compared_value(op, value)

    def process_bind_param(self, value: Any, _dialect: sa.Dialect) -> Any:
        """Convert a Python value into a JSON string for the database."""
        if value is not None:
            value = json.dumps(value, default=str)  # Callable default
        return value

    def process_result_value(self, value: Any, _dialect: sa.Dialect) -> Any:
        """Convert a JSON string from the database into a dict."""
        if value is not None and isinstance(value, str):
            # Psycopg2 >= 2.5 will auto-decode JSON columns, so
            # we only attempt decoding if the value is a string.
            # Since this column stores dicts only, processed values
            # can never be strings.
            value = json.loads(value)
        return value


class MutableDict(Mutable, dict):
    @classmethod
    def coerce(cls, _key: Any, value: Any) -> Optional[MutableDict]:
        """Convert plain dictionaries to MutableDict."""
        if value is None:
            return None
        if not isinstance(value, MutableDict):
            if isinstance(value, Mapping):
                return MutableDict(value)
            if isinstance(value, str):
                # Got a string, attempt to parse as JSON
                try:
                    return MutableDict(json.loads(value))
                except ValueError:
                    raise ValueError(f"Invalid JSON string: {value!r}") from None
            raise ValueError(f"Value is not dict-like: {value!r}")
        return value

    def __setitem__(self, key: Any, value: Any) -> None:
        """Detect dictionary set events and emit change events."""
        dict.__setitem__(self, key, value)
        self.changed()

    def __delitem__(self, key: Any) -> None:
        """Detect dictionary del events and emit change events."""
        dict.__delitem__(self, key)
        self.changed()


MutableDict.associate_with(JsonDict)


class UrlType(UrlTypeBase):
    """
    Extension of URLType_ from SQLAlchemy-Utils that ensures URLs are well formed.

    .. _URLType: https://sqlalchemy-utils.readthedocs.io/en/latest/data_types.html#module-sqlalchemy_utils.types.url

    :param schemes: Valid URL schemes. Use `None` to allow any scheme,
        `()` for no scheme
    :param optional_scheme: Schemes are optional (allows URLs starting with ``//``)
    :param optional_host: Allow URLs without a hostname (required for ``mailto`` and
        ``file`` schemes)
    """

    impl = sa.Unicode
    url_parser = furl
    cache_ok = True

    def __init__(
        self,
        schemes: Optional[Collection[str]] = ('http', 'https'),
        optional_scheme: bool = False,
        optional_host: bool = False,
    ) -> None:
        super().__init__()
        self.schemes = schemes
        self.optional_host = optional_host
        self.optional_scheme = optional_scheme

    def process_bind_param(self, value: Any, dialect: sa.Dialect) -> Optional[str]:
        """Validate URL before storing to the database."""
        value = super().process_bind_param(value, dialect)
        if value:
            parsed = self.url_parser(value)
            # If scheme is present, it must be valid
            # If not present, the optional flag must be True
            if parsed.scheme:
                if self.schemes is not None and parsed.scheme not in self.schemes:
                    raise ValueError("Invalid URL scheme")
            elif not self.optional_scheme:
                raise ValueError("Missing URL scheme")

            # Host may be missing only if optional
            if not parsed.host and not self.optional_host:
                raise ValueError("Missing URL host")
        return value

    def process_result_value(self, value: Any, _dialect: sa.Dialect) -> Optional[furl]:
        """Cast URL loaded from database into a furl object."""
        if value is not None:
            return self.url_parser(value)
        return None

    def _coerce(self, value: Any) -> Optional[furl]:
        if value is not None and not isinstance(value, self.url_parser):
            return self.url_parser(value)
        return value

1	"""SQLAlchemy column types."""	4✔
2
3	from __future__ import annotations	4✔
4
5	import json	4✔
6	from collections.abc import Collection, Mapping	4✔
7	from typing import Any, Optional	4✔
8
9	import sqlalchemy as sa	4✔
10	from furl import furl	4✔
11	from sqlalchemy.dialects import postgresql	4✔
12	from sqlalchemy.ext.mutable import Mutable	4✔
13	from sqlalchemy.types import TypeDecorator	4✔
14	from sqlalchemy_utils.types import URLType as UrlTypeBase	4✔
15
16	__all__ = ['JsonDict', 'UrlType']	4✔
17
18
19	# Adapted from http://docs.sqlalchemy.org/en/rel_0_8/orm/extensions/mutable.html
20	# #establishing-mutability-on-scalar-column-values
21
22
23	class JsonDict(TypeDecorator):	4✔
24	"""	4✔
25	Represents a JSON data structure.
26
27	Usage::
28
29	column = Column(JsonDict)
30
31	The column will be represented to the database as a ``JSONB`` column if
32	the server is PostgreSQL 9.4 or later, ``JSON`` if PostgreSQL 9.2 or 9.3,
33	and ``TEXT`` for everything else. The column behaves like a JSON store
34	regardless of the backing data type.
35	"""
36
37	# TypeDecorator replaces the Type with an instance of the type in the instance
38	impl: sa.types.JSON = sa.types.JSON # type: ignore[assignment]	4✔
39	cache_ok = False	4✔
40
41	def load_dialect_impl(self, dialect: sa.Dialect) -> sa.types.TypeEngine:	4✔
42	"""Use JSONB column in PostgreSQL."""
43	if dialect.name == 'postgresql':	4✔
44	return dialect.type_descriptor(postgresql.JSONB) # type: ignore[arg-type]	4✔
UNCOV 45	return dialect.type_descriptor(self.impl)	×
46
47	def coerce_compared_value(self, op: Any, value: Any) -> sa.types.TypeEngine:	4✔
48	"""Coerce an incoming value using the JSON type's default handler."""
UNCOV 49	return self.impl.coerce_compared_value(op, value)	×
50
51	def process_bind_param(self, value: Any, _dialect: sa.Dialect) -> Any:	4✔
52	"""Convert a Python value into a JSON string for the database."""
53	if value is not None:	4✔
54	value = json.dumps(value, default=str) # Callable default	4✔
55	return value	4✔
56
57	def process_result_value(self, value: Any, _dialect: sa.Dialect) -> Any:	4✔
58	"""Convert a JSON string from the database into a dict."""
59	if value is not None and isinstance(value, str):	4✔
60	# Psycopg2 >= 2.5 will auto-decode JSON columns, so
61	# we only attempt decoding if the value is a string.
62	# Since this column stores dicts only, processed values
63	# can never be strings.
64	value = json.loads(value)	4✔
65	return value	4✔
66
67
68	class MutableDict(Mutable, dict):	4✔
69	@classmethod	4✔
70	def coerce(cls, _key: Any, value: Any) -> Optional[MutableDict]:	4✔
71	"""Convert plain dictionaries to MutableDict."""
72	if value is None:	4✔
73	return None	4✔
74	if not isinstance(value, MutableDict):	4✔
75	if isinstance(value, Mapping):	4✔
76	return MutableDict(value)	4✔
77	if isinstance(value, str):	4✔
78	# Got a string, attempt to parse as JSON
79	try:	4✔
80	return MutableDict(json.loads(value))	4✔
81	except ValueError:	4✔
82	raise ValueError(f"Invalid JSON string: {value!r}") from None	4✔
83	raise ValueError(f"Value is not dict-like: {value!r}")	4✔
UNCOV 84	return value	×
85
86	def __setitem__(self, key: Any, value: Any) -> None:	4✔
87	"""Detect dictionary set events and emit change events."""
88	dict.__setitem__(self, key, value)	4✔
89	self.changed()	4✔
90
91	def __delitem__(self, key: Any) -> None:	4✔
92	"""Detect dictionary del events and emit change events."""
93	dict.__delitem__(self, key)	4✔
94	self.changed()	4✔
95
96
97	MutableDict.associate_with(JsonDict)	4✔
98
99
100	class UrlType(UrlTypeBase):	4✔
101	"""	4✔
102	Extension of URLType_ from SQLAlchemy-Utils that ensures URLs are well formed.
103
104	.. _URLType: https://sqlalchemy-utils.readthedocs.io/en/latest/data_types.html#module-sqlalchemy_utils.types.url
105
106	:param schemes: Valid URL schemes. Use `None` to allow any scheme,
107	`()` for no scheme
108	:param optional_scheme: Schemes are optional (allows URLs starting with ``//``)
109	:param optional_host: Allow URLs without a hostname (required for ``mailto`` and
110	``file`` schemes)
111	"""
112
113	impl = sa.Unicode	4✔
114	url_parser = furl	4✔
115	cache_ok = True	4✔
116
117	def __init__(	4✔
118	self,
119	schemes: Optional[Collection[str]] = ('http', 'https'),
120	optional_scheme: bool = False,
121	optional_host: bool = False,
122	) -> None:
123	super().__init__()	4✔
124	self.schemes = schemes	4✔
125	self.optional_host = optional_host	4✔
126	self.optional_scheme = optional_scheme	4✔
127
128	def process_bind_param(self, value: Any, dialect: sa.Dialect) -> Optional[str]:	4✔
129	"""Validate URL before storing to the database."""
130	value = super().process_bind_param(value, dialect)	4✔
131	if value:	4✔
132	parsed = self.url_parser(value)	4✔
133	# If scheme is present, it must be valid
134	# If not present, the optional flag must be True
135	if parsed.scheme:	4✔
136	if self.schemes is not None and parsed.scheme not in self.schemes:	4✔
137	raise ValueError("Invalid URL scheme")	4✔
138	elif not self.optional_scheme:	4✔
139	raise ValueError("Missing URL scheme")	4✔
140
141	# Host may be missing only if optional
142	if not parsed.host and not self.optional_host:	4✔
143	raise ValueError("Missing URL host")	4✔
144	return value	4✔
145
146	def process_result_value(self, value: Any, _dialect: sa.Dialect) -> Optional[furl]:	4✔
147	"""Cast URL loaded from database into a furl object."""
148	if value is not None:	4✔
149	return self.url_parser(value)	4✔
150	return None	4✔
151
152	def _coerce(self, value: Any) -> Optional[furl]:	4✔
UNCOV 153	if value is not None and not isinstance(value, self.url_parser):	×
154	return self.url_parser(value)	×
155	return value	×

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