Function auth_login

app/routers/auth.py:142–269

(
    request: Request,
    settings: Settings = Depends(get_settings),
    db: AsyncSession = Depends(get_db),
    redis=Depends(get_redis_client),
)

@router.api_route("/login", methods=["GET", "POST"], name="auth_login")
async def auth_login(
    request: Request,
    settings: Settings = Depends(get_settings),
    db: AsyncSession = Depends(get_db),
    redis=Depends(get_redis_client),
):
    try:
        current_user = await get_current_user(request, db, settings)
        if current_user:
            return RedirectResponse("/", status_code=303)
    except HTTPException:
        pass

    form: Any = await EmailLoginForm.from_formdata(request)

    if request.method == "POST" and await form.validate_on_submit():
        email = form.email.data
        if not await is_email_allowed(email, db):
            await notify_denied(
                email,
                "email",
                request,
                settings.access_denied_webhook,
            )
            flash(request, _(settings.access_denied_message), "error")
            return RedirectResponseX(
                request.url_for("auth_login"),
                status_code=303,
                request=request,
            )
        expires_at = utc_now() + timedelta(seconds=settings.magic_link_ttl_seconds)
        jti = secrets.token_urlsafe(32)
        token_payload = {
            "email": email,
            "exp": int(expires_at.timestamp()),
            "iat": int(utc_now().timestamp()),
            "type": "email_login",
            "jti": jti,
        }
        magic_token = jwt.encode({"alg": "HS256"}, token_payload, settings.secret_key)
        magic_token_str = (
            magic_token.decode("utf-8")
            if isinstance(magic_token, bytes)
            else magic_token
        )

        verify_link = str(
            request.url_for("auth_email_verify").include_query_params(
                token=magic_token_str
            )
        )

        try:
            await redis.setex(
                f"magic_link:email_login:{jti}",
                settings.magic_link_ttl_seconds,
                "1",
            )