⚡ Bolt: Optimize single-model database fetches

src/h4ckath0n/auth/passkeys/service.py

@@ -201,13 +201,14 @@ async def finish_authentication(
     flow = await _get_valid_flow(db, flow_id, "authenticate")
 
     raw_id = credential_json.get("rawId") or credential_json.get("id", "")
-    result = await db.execute(
+    # Optimization: Use db.scalar to avoid intermediate ExecutionResult allocation.
+    stored = await db.scalar(
         select(WebAuthnCredential).filter(
             WebAuthnCredential.credential_id == raw_id,
             WebAuthnCredential.revoked_at.is_(None),
         )
     )
-    if (stored := result.scalars().first()) is None:
+    if stored is None:
         raise ValueError("Unknown or revoked credential")
 
     challenge_bytes = base64url_to_bytes(flow.challenge)

src/h4ckath0n/auth/service.py

@@ -11,7 +11,7 @@
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from sqlalchemy import func, select
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from h4ckath0n.auth.models import Device, PasswordResetToken, User
@@ -40,12 +40,11 @@ async def _is_bootstrap_admin(email: str, settings: Settings, db: AsyncSession)
     """Decide whether a newly-registered user should be admin."""
     if email in settings.bootstrap_admin_emails:
         return True
-    if settings.first_user_is_admin:
-        result = await db.execute(select(func.count()).select_from(User))
-        count = result.scalar()
-        if count == 0:
-            return True
-    return False
+    # Optimization: Use limit(1) instead of count() to quickly check if any user exists.
+    return bool(
+        settings.first_user_is_admin
+        and (await db.scalar(select(User.id).limit(1))) is None
+    )
 
 
 async def register_user(
@@ -57,8 +56,8 @@ async def register_user(
     display_name: str | None = None,
 ) -> User:
     hash_password, _verify = _require_password_extra()
-    result = await db.execute(select(User).filter(User.email == email))
-    if result.scalars().first():
+    # Optimization: Use db.scalar for existence check to avoid ORM parsing.
+    if await db.scalar(select(User.id).filter(User.email == email)):
         raise ValueError("Email already registered")
     role = "admin" if await _is_bootstrap_admin(email, settings, db) else "user"
     user = User(
@@ -82,8 +81,8 @@ async def register_user(
 
 async def authenticate_user(db: AsyncSession, email: str, password: str) -> User | None:
     _hash, verify_password = _require_password_extra()
-    result = await db.execute(select(User).filter(User.email == email))
-    user = result.scalars().first()
+    # Optimization: Use db.scalar to avoid intermediate ExecutionResult allocation.
+    user = await db.scalar(select(User).filter(User.email == email))
     if user is None or not user.password_hash:
         verify_password(password, _DUMMY_PASSWORD_HASH)
         return None
@@ -172,13 +171,14 @@ async def confirm_password_reset(
     """Confirm a password reset and return the user."""
     hash_password, _verify = _require_password_extra()
     hashed = _hash_token(raw_token)
-    prt_result = await db.execute(
+    # Optimization: Use db.scalar to avoid intermediate ExecutionResult allocation.
+    prt = await db.scalar(
         select(PasswordResetToken).filter(
             PasswordResetToken.token_hash == hashed,
             PasswordResetToken.used.is_(False),
         )
     )
-    if (prt := prt_result.scalars().first()) is None:
+    if prt is None:
         raise ValueError("Invalid or already-used reset token")
     if prt.expires_at.replace(tzinfo=UTC) < datetime.now(UTC):
         raise ValueError("Reset token expired")

src/h4ckath0n/cli/_common.py

@@ -154,7 +154,8 @@ def _resolve_user(session: Session, args: argparse.Namespace) -> User | None:
         return session.get(User, user_id)
 
     stmt = select(User).where(User.email == email)
-    return session.execute(stmt).scalars().first()
+    # Optimization: Use session.scalar to avoid intermediate ExecutionResult allocation.
+    return session.scalar(stmt)
 
 
 def _user_or_exit(