0.12.0a6 add user_loader()

CHANGELOG.md

@@ -6,6 +6,7 @@
 * New module `mat` adds a shallow reimplementation of `Matrix()` in order to implement matrix multiplication
 * Removed obsolete `configparse` implementation that has been around since 0.3 and shelved since 0.4.
 * `color`: added support for conversion from RGB to sRGB, XYZ, OKLab and OKLCH.
+* Added `user-loader` for Quart-Auth and SQLAlchemy
 
 ## 0.11.2
 

pyproject.toml

@@ -39,7 +39,7 @@ Documentation = "https://suou.readthedocs.io"
 # the below are all dev dependencies (and probably already installed)
 sqlalchemy = [
     "SQLAlchemy[asyncio]>=2.0.0",
-    "flask-sqlalchemy"
+    "flask-sqlalchemy"  # glue code
 ]
 flask = [
     "Flask>=2.0.0",
@@ -61,6 +61,10 @@ quart = [
     "Quart-Schema",
     "starlette>=0.47.2"
 ]
+quart_auth = [
+    "Quart-Auth",
+    "suou[sqlalchemy]"  # glue code
+]
 sass = [
     ## HEADS UP!! libsass carries a C extension + uses setup.py
     "libsass"
@@ -70,6 +74,7 @@ full = [
     "suou[sqlalchemy]",
     "suou[flask]",
     "suou[quart]",
+    "suou[quart_auth]",
     "suou[peewee]",
     "suou[markdown]",
     "suou[sass]"

src/suou/__init__.py

@@ -38,7 +38,7 @@ from .http import WantsContentType
 from .color import OKLabColor, chalk, WebColor, RGBColor, SRGBColor, XYZColor, OKLabColor
 from .mat import Matrix
 
-__version__ = "0.12.0a5"
+__version__ = "0.12.0a6"
 
 __all__ = (
     'ConfigOptions', 'ConfigParserConfigSource', 'ConfigSource', 'ConfigValue',

src/suou/quart_auth.py

@@ -0,0 +1,97 @@
+"""
+Utilities for Quart-Auth
+
+---
+
+Copyright (c) 2025 Sakuragasaki46.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+See LICENSE for the specific language governing permissions and
+limitations under the License.
+
+This software is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+"""
+
+from __future__ import annotations
+from typing import Callable, TypeVar
+from quart_auth import AuthUser, Action
+from sqlalchemy import select
+from sqlalchemy.orm import DeclarativeBase
+from .sqlalchemy.asyncio import AsyncSession, SQLAlchemy
+
+_T = TypeVar('_T')
+
+def user_loader(database: SQLAlchemy, user_class: type[DeclarativeBase], *,
+        attr_loader: Callable[[type[AuthUser], str], _T] = lambda x, y: x.id == int(y)
+    ):
+    """
+    Returns a properly subclassed AuthUser loader for use in Quart-Auth.
+
+    Actual User object is at .user; other attributes are proxied.
+
+    Requires to be awaited before request before usage.
+
+    Uses SQLAlchemy's AsyncSession.
+
+    Parameters:
+    * database The database instance.
+    * user_class The user class.
+    * attr_loader A lambda taking user_class and auth_id, default (user_class, auth_id : user_class.id == int(auth_id))
+
+    *New in 0.12.0*
+    """
+    class UserLoader(AuthUser):
+        _auth_id: str | None
+        _auth_obj: user_class | None
+        id: _T
+
+        def __init__(self, auth_id: str | None, action: Action = Action.PASS):
+            self._auth_id = auth_id
+            self._auth_obj = None
+            self._auth_sess: AsyncSession | None = None
+            self.action = action
+
+        @property
+        def auth_id(self) -> str | None:
+            return self._auth_id
+
+        @property
+        async def is_authenticated(self) -> bool:
+            await self._load()
+            return self._auth_id is not None
+        
+        async def _load(self):
+            if self._auth_obj is None and self._auth_id is not None:
+                async with database as session:
+                    self._auth_obj = (await session.execute(select(user_class).where(attr_loader(user_class, self._auth_id)))).scalar()
+                    if self._auth_obj is None:
+                        raise RuntimeError('failed to fetch user')
+        
+        def __getattr__(self, key):
+            if self._auth_obj is None:
+                raise RuntimeError('user is not loaded')
+            return getattr(self._auth_obj, key)
+
+        def __bool__(self):
+            return self._auth_obj is not None
+
+        @property
+        def session(self):
+            return self._auth_sess
+
+        async def _unload(self):
+            # user is not expected to mutate
+            if self._auth_sess:
+                await self._auth_sess.rollback()
+        
+        @property
+        def user(self):
+            return self._auth_obj
+
+    return UserLoader
+
+# Optional dependency: do not import into __init__.py
+__all__ = ('user_loader',)
+