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

94.29

/src/coaster/sqlalchemy/annotations.py

"""
SQLAlchemy attribute annotations.

Annotations are strings attached to attributes that serve as a programmer
reference on how those attributes are meant to be used. They can be used to
indicate that a column's value should be :attr:`immutable` and should never
change, or that it's a :attr:`cached` copy of a value from another source
that can be safely discarded in case of a conflict.

This module's exports may be imported via :mod:`coaster.sqlalchemy`.

Sample usage::

    import sqlalchemy as sa
    from sqlalchemy.orm import Mapped, mapped_column
    from coaster.sqlalchemy import annotation_wrapper, immutable
    from . import Model

    natural_key = annotation_wrapper('natural_key', "Natural key for this model")


    class MyModel(Model):
        __tablename__ = 'my_model'
        id: Mapped[int] = immutable(mapped_column(sa.Integer, primary_key=True))
        name: Mapped[str] = natural_key(mapped_column(sa.Unicode(250), unique=True))

        @classmethod
        def get(cls, **kwargs):
            for key in kwargs:
                if key in cls.__column_annotations__[natural_key.name]:
                    return cls.query.filter_by(**{key: kwargs[key]}).one_or_none()

Annotations are saved to the model's class as a ``__column_annotations__``
dictionary, mapping annotation names to a list of attribute names, and to a
reverse lookup ``__column_annotations_by_attr__`` of attribute names to annotations.

.. deprecated:: 0.7.0
    This module is due to be replaced with typing.Annotated
"""

from __future__ import annotations

from collections.abc import Hashable
from typing import Any, Callable, Optional, TypeVar

import sqlalchemy as sa
from sqlalchemy.orm import (
    ColumnProperty,
    MappedColumn,
    Mapper,
    MapperProperty,
    RelationshipProperty,
    SynonymProperty,
)
from sqlalchemy.orm.attributes import QueryableAttribute
from sqlalchemy.schema import SchemaItem

from ..signals import coaster_signals

__all__ = ['annotations_configured', 'annotation_wrapper']

# Global dictionary for temporary storage of annotations until the
# mapper_configured events
__cache__: dict[Any, list] = {}

# --- Constructor ----------------------------------------------------------------------


_A = TypeVar('_A', bound=Any)


def annotation_wrapper(
    annotation: str, doc: Optional[str] = None
) -> Callable[[_A], _A]:
    """Define an annotation, which can be applied to attributes in a database model."""

    def decorator(attr: _A) -> _A:
        __cache__.setdefault(attr, []).append(annotation)
        # Also mark the annotation on the object itself. This will
        # fail if the object has a restrictive __slots__, but it's
        # required for some objects like Column because SQLAlchemy copies
        # them in subclasses, changing their hash and making them
        # undiscoverable via the cache.
        if isinstance(attr, SynonymProperty):
            raise TypeError(
                "Synonyms cannot have annotations; set it on the referred attribute"
            )
        if isinstance(attr, MappedColumn):
            # pylint: disable=protected-access
            if not hasattr(attr.column, '_coaster_annotations'):
                attr.column._coaster_annotations = []  # type: ignore[attr-defined]
            attr.column._coaster_annotations.append(annotation)

        if isinstance(attr, (SchemaItem, ColumnProperty, MapperProperty)):
            attr.info.setdefault('_coaster_annotations', []).append(annotation)
        else:
            try:
                # pylint: disable=protected-access
                if not hasattr(attr, '_coaster_annotations'):
                    attr._coaster_annotations = []
                attr._coaster_annotations.append(annotation)
            except AttributeError:
                pass
        return attr

    decorator.__name__ = annotation
    decorator.__doc__ = doc
    return decorator


# --- Signals --------------------------------------------------------------------------

annotations_configured = coaster_signals.signal(
    'annotations-configured',
    doc="Signal raised after all annotations on a class are configured",
)


# --- Annotation processor -------------------------------------------------------------


@sa.event.listens_for(Mapper, 'mapper_configured')
def _configure_annotations(_mapper: Any, cls: type[Any]) -> None:
    """
    Extract annotations from attributes.

    Run through attributes of the class looking for annotations from
    :func:`annotation_wrapper` and add them to :attr:`cls.__column_annotations__`
    and :attr:`cls.__column_annotations_by_attr__`
    """
    annotations: dict[str, list[str]] = {}  # Annotation name: list of attrs
    annotations_by_attr: dict[str, list[str]] = {}  # Attr name: annotations

    # An attribute may be defined more than once in base classes. Only handle the first
    processed = set()

    # Loop through all attributes in the class and its base classes,
    # looking for annotations
    for base in cls.__mro__:
        for name, attr in base.__dict__.items():
            # pylint: disable=protected-access
            if name in processed or name.startswith('__'):
                continue

            if isinstance(attr, Hashable) and attr in __cache__:
                data = __cache__[attr]
            elif isinstance(attr, QueryableAttribute) and isinstance(
                getattr(attr, 'original_property', None), SynonymProperty
            ):
                # Skip synonyms
                data = None
            # 'data' is a list of string annotations
            elif isinstance(attr, MappedColumn) and hasattr(
                attr.column, '_coaster_annotations'
            ):
                data = attr.column._coaster_annotations
            elif hasattr(attr, '_coaster_annotations'):
                # pylint: disable=protected-access
                data = attr._coaster_annotations
            elif isinstance(
                attr, (QueryableAttribute, RelationshipProperty, MapperProperty)
            ):
                if attr.property in __cache__:
                    data = __cache__[attr.property]
                elif '_coaster_annotations' in attr.info:
                    data = attr.info['_coaster_annotations']
                elif hasattr(attr.property, '_coaster_annotations'):
                    # pylint: disable=protected-access
                    data = attr.property._coaster_annotations
                else:
                    data = None
            else:
                data = None
            if data is not None:
                annotations_by_attr.setdefault(name, []).extend(data)
                for a in data:
                    annotations.setdefault(a, []).append(name)
                processed.add(name)

    # Classes specifying ``__column_annotations__`` directly isn't supported,
    # so we don't bother preserving existing content, if any.
    if annotations:
        cls.__column_annotations__ = annotations
    if annotations_by_attr:
        cls.__column_annotations_by_attr__ = annotations_by_attr
    annotations_configured.send(cls)

1	"""	4✔
2	SQLAlchemy attribute annotations.
3
4	Annotations are strings attached to attributes that serve as a programmer
5	reference on how those attributes are meant to be used. They can be used to
6	indicate that a column's value should be :attr:`immutable` and should never
7	change, or that it's a :attr:`cached` copy of a value from another source
8	that can be safely discarded in case of a conflict.
9
10	This module's exports may be imported via :mod:`coaster.sqlalchemy`.
11
12	Sample usage::
13
14	import sqlalchemy as sa
15	from sqlalchemy.orm import Mapped, mapped_column
16	from coaster.sqlalchemy import annotation_wrapper, immutable
17	from . import Model
18
19	natural_key = annotation_wrapper('natural_key', "Natural key for this model")
20
21
22	class MyModel(Model):
23	__tablename__ = 'my_model'
24	id: Mapped[int] = immutable(mapped_column(sa.Integer, primary_key=True))
25	name: Mapped[str] = natural_key(mapped_column(sa.Unicode(250), unique=True))
26
27	@classmethod
28	def get(cls, **kwargs):
29	for key in kwargs:
30	if key in cls.__column_annotations__[natural_key.name]:
31	return cls.query.filter_by(**{key: kwargs[key]}).one_or_none()
32
33	Annotations are saved to the model's class as a ``__column_annotations__``
34	dictionary, mapping annotation names to a list of attribute names, and to a
35	reverse lookup ``__column_annotations_by_attr__`` of attribute names to annotations.
36
37	.. deprecated:: 0.7.0
38	This module is due to be replaced with typing.Annotated
39	"""
40
41	from __future__ import annotations	4✔
42
43	from collections.abc import Hashable	4✔
44	from typing import Any, Callable, Optional, TypeVar	4✔
45
46	import sqlalchemy as sa	4✔
47	from sqlalchemy.orm import (	4✔
48	ColumnProperty,
49	MappedColumn,
50	Mapper,
51	MapperProperty,
52	RelationshipProperty,
53	SynonymProperty,
54	)
55	from sqlalchemy.orm.attributes import QueryableAttribute	4✔
56	from sqlalchemy.schema import SchemaItem	4✔
57
58	from ..signals import coaster_signals	4✔
59
60	__all__ = ['annotations_configured', 'annotation_wrapper']	4✔
61
62	# Global dictionary for temporary storage of annotations until the
63	# mapper_configured events
64	__cache__: dict[Any, list] = {}	4✔
65
66	# --- Constructor ----------------------------------------------------------------------
67
68
69	_A = TypeVar('_A', bound=Any)	4✔
70
71
72	def annotation_wrapper(	4✔
73	annotation: str, doc: Optional[str] = None
74	) -> Callable[[_A], _A]:
75	"""Define an annotation, which can be applied to attributes in a database model."""
76
77	def decorator(attr: _A) -> _A:	4✔
78	__cache__.setdefault(attr, []).append(annotation)	4✔
79	# Also mark the annotation on the object itself. This will
80	# fail if the object has a restrictive __slots__, but it's
81	# required for some objects like Column because SQLAlchemy copies
82	# them in subclasses, changing their hash and making them
83	# undiscoverable via the cache.
84	if isinstance(attr, SynonymProperty):	4✔
UNCOV 85	raise TypeError(	×
86	"Synonyms cannot have annotations; set it on the referred attribute"
87	)
88	if isinstance(attr, MappedColumn):	4✔
89	# pylint: disable=protected-access
90	if not hasattr(attr.column, '_coaster_annotations'):	4✔
91	attr.column._coaster_annotations = [] # type: ignore[attr-defined]	4✔
92	attr.column._coaster_annotations.append(annotation)	4✔
93
94	if isinstance(attr, (SchemaItem, ColumnProperty, MapperProperty)):	4✔
95	attr.info.setdefault('_coaster_annotations', []).append(annotation)	4✔
96	else:
97	try:	4✔
98	# pylint: disable=protected-access
99	if not hasattr(attr, '_coaster_annotations'):	4✔
100	attr._coaster_annotations = []	4✔
101	attr._coaster_annotations.append(annotation)	4✔
102	except AttributeError:	4✔
103	pass	4✔
104	return attr	4✔
105
106	decorator.__name__ = annotation	4✔
107	decorator.__doc__ = doc	4✔
108	return decorator	4✔
109
110
111	# --- Signals --------------------------------------------------------------------------
112
113	annotations_configured = coaster_signals.signal(	4✔
114	'annotations-configured',
115	doc="Signal raised after all annotations on a class are configured",
116	)
117
118
119	# --- Annotation processor -------------------------------------------------------------
120
121
122	@sa.event.listens_for(Mapper, 'mapper_configured')	4✔
123	def _configure_annotations(_mapper: Any, cls: type[Any]) -> None:	4✔
124	"""
125	Extract annotations from attributes.
126
127	Run through attributes of the class looking for annotations from
128	:func:`annotation_wrapper` and add them to :attr:`cls.__column_annotations__`
129	and :attr:`cls.__column_annotations_by_attr__`
130	"""
131	annotations: dict[str, list[str]] = {} # Annotation name: list of attrs	4✔
132	annotations_by_attr: dict[str, list[str]] = {} # Attr name: annotations	4✔
133
134	# An attribute may be defined more than once in base classes. Only handle the first
135	processed = set()	4✔
136
137	# Loop through all attributes in the class and its base classes,
138	# looking for annotations
139	for base in cls.__mro__:	4✔
140	for name, attr in base.__dict__.items():	4✔
141	# pylint: disable=protected-access
142	if name in processed or name.startswith('__'):	4✔
143	continue	4✔
144
145	if isinstance(attr, Hashable) and attr in __cache__:	4✔
146	data = __cache__[attr]	4✔
147	elif isinstance(attr, QueryableAttribute) and isinstance(	4✔
148	getattr(attr, 'original_property', None), SynonymProperty
149	):
150	# Skip synonyms
151	data = None	4✔
152	# 'data' is a list of string annotations
153	elif isinstance(attr, MappedColumn) and hasattr(	4✔
154	attr.column, '_coaster_annotations'
155	):
UNCOV 156	data = attr.column._coaster_annotations	×
157	elif hasattr(attr, '_coaster_annotations'):	4✔
158	# pylint: disable=protected-access
159	data = attr._coaster_annotations	4✔
160	elif isinstance(	4✔
161	attr, (QueryableAttribute, RelationshipProperty, MapperProperty)
162	):
163	if attr.property in __cache__:	4✔
164	data = __cache__[attr.property]	4✔
165	elif '_coaster_annotations' in attr.info:	4✔
UNCOV 166	data = attr.info['_coaster_annotations']	×
167	elif hasattr(attr.property, '_coaster_annotations'):	4✔
168	# pylint: disable=protected-access
UNCOV 169	data = attr.property._coaster_annotations	×
170	else:
171	data = None	4✔
172	else:
173	data = None	4✔
174	if data is not None:	4✔
175	annotations_by_attr.setdefault(name, []).extend(data)	4✔
176	for a in data:	4✔
177	annotations.setdefault(a, []).append(name)	4✔
178	processed.add(name)	4✔
179
180	# Classes specifying ``__column_annotations__`` directly isn't supported,
181	# so we don't bother preserving existing content, if any.
182	if annotations:	4✔
183	cls.__column_annotations__ = annotations	4✔
184	if annotations_by_attr:	4✔
185	cls.__column_annotations_by_attr__ = annotations_by_attr	4✔
186	annotations_configured.send(cls)	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