yusur/suou
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

0.7.6 fix @glue() stray usage

Browse source
This commit is contained in:
Yusur 2025-11-01 10:05:22 +01:00
parent 556019e0bd
commit 96a65c38e3
3 changed files with 62 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

9
aliases/sakuragasaki46_suou/pyproject.toml
View file

@ -10,7 +10,7 @@ license = "Apache-2.0"
readme = "README.md"
dependencies = [
"suou==0.7.4",
"suou==0.7.5",
"itsdangerous",
"toml",
"pydantic",
@ -39,14 +39,16 @@ Documentation = "https://suou.readthedocs.io"
[project.optional-dependencies]
# the below are all dev dependencies (and probably already installed)
sqlalchemy = [
"SQLAlchemy[asyncio]>=2.0.0"
"SQLAlchemy[asyncio]>=2.0.0",
"flask-sqlalchemy"
]
flask = [
"Flask>=2.0.0",
"Flask-RestX"
]
flask_sqlalchemy = [
"Flask-SqlAlchemy",
"sakuragasaki46_suou[sqlalchemy]",
"sakuragasaki46_suou[flask]"
]
peewee = [
## HEADS UP! peewee has setup.py, may slow down installation
@ -71,7 +73,6 @@ full = [
"sakuragasaki46_suou[quart]",
"sakuragasaki46_suou[peewee]",
"sakuragasaki46_suou[markdown]",
"sakuragasaki46_suou[flask-sqlalchemy]",
"sakuragasaki46_suou[sass]"
]

2
src/suou/__init__.py
View file

@ -37,7 +37,7 @@ from .redact import redact_url_password
from .http import WantsContentType
from .color import chalk
__version__ = "0.7.5"
__version__ = "0.7.6"
__all__ = (
'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',

114
src/suou/sqlalchemy/asyncio.py
View file

@ -119,76 +119,72 @@ class SQLAlchemy:
# XXX NOT public API! DO NOT USE
current_session: ContextVar[AsyncSession] = ContextVar('current_session')
## experimental
@glue('flask_sqlalchemy')
def _make_AsyncSelectPagination(flask_sqlalchemy):
class AsyncSelectPagination(flask_sqlalchemy.pagination.Pagination):
"""
flask_sqlalchemy.SelectPagination but asynchronous.
Pagination is not part of the public API, therefore expect that it may break
"""
async def _query_items(self) -> list:
select_q: Select = self._query_args["select"]
select = select_q.limit(self.per_page).offset(self._query_offset)
session: AsyncSession = self._query_args["session"]
out = (await session.execute(select)).scalars()
return out
async def _query_count(self) -> int:
select_q: Select = self._query_args["select"]
sub = select_q.options(lazyload("*")).order_by(None).subquery()
session: AsyncSession = self._query_args["session"]
out = (await session.execute(select(func.count()).select_from(sub))).scalar()
return out
class AsyncSelectPagination(flask_sqlalchemy.pagination.Pagination):
"""
flask_sqlalchemy.SelectPagination but asynchronous.
def __init__(self,
page: int | None = None,
per_page: int | None = None,
max_per_page: int | None = 100,
error_out: Exception | None = NotFoundError,
count: bool = True,
**kwargs):
## XXX flask-sqlalchemy says Pagination() is not public API.
## Things may break; beware.
self._query_args = kwargs
page, per_page = self._prepare_page_args(
page=page,
per_page=per_page,
max_per_page=max_per_page,
error_out=error_out,
)
Pagination is not part of the public API, therefore expect that it may break
"""
self.page: int = page
"""The current page."""
async def _query_items(self) -> list:
select_q: Select = self._query_args["select"]
select = select_q.limit(self.per_page).offset(self._query_offset)
session: AsyncSession = self._query_args["session"]
out = (await session.execute(select)).scalars()
return out
self.per_page: int = per_page
"""The maximum number of items on a page."""
async def _query_count(self) -> int:
select_q: Select = self._query_args["select"]
sub = select_q.options(lazyload("*")).order_by(None).subquery()
session: AsyncSession = self._query_args["session"]
out = (await session.execute(select(func.count()).select_from(sub))).scalar()
return out
self.max_per_page: int | None = max_per_page
"""The maximum allowed value for ``per_page``."""
def __init__(self,
page: int | None = None,
per_page: int | None = None,
max_per_page: int | None = 100,
error_out: Exception | None = NotFoundError,
count: bool = True,
**kwargs):
## XXX flask-sqlalchemy says Pagination() is not public API.
## Things may break; beware.
self._query_args = kwargs
page, per_page = self._prepare_page_args(
page=page,
per_page=per_page,
max_per_page=max_per_page,
error_out=error_out,
)
self.items = None
self.total = None
self.error_out = error_out
self.has_count = count
self.page: int = page
"""The current page."""
async def __aiter__(self):
self.items = await self._query_items()
if self.items is None:
raise RuntimeError('query returned None')
if not self.items and self.page != 1 and self.error_out:
raise self.error_out
if self.has_count:
self.total = await self._query_count()
for i in self.items:
yield i
self.per_page: int = per_page
"""The maximum number of items on a page."""
return AsyncSelectPagination
self.max_per_page: int | None = max_per_page
"""The maximum allowed value for ``per_page``."""
self.items = None
self.total = None
self.error_out = error_out
self.has_count = count
async def __aiter__(self):
self.items = await self._query_items()
if self.items is None:
raise RuntimeError('query returned None')
if not self.items and self.page != 1 and self.error_out:
raise self.error_out
if self.has_count:
self.total = await self._query_count()
for i in self.items:
yield i
AsyncSelectPagination = _make_AsyncSelectPagination()
del _make_AsyncSelectPagination
def async_query(db: SQLAlchemy, multi: False):