removing flask_login, fixed many appwriter issues with custom class
This commit is contained in:
@@ -1,16 +1,17 @@
|
||||
import os
|
||||
from flask import Flask
|
||||
from flask import Flask, redirect, url_for, request, g, session, flash
|
||||
from dotenv import load_dotenv
|
||||
from .utils.extensions import login_manager
|
||||
from .blueprints.auth.routes import auth_bp
|
||||
from .blueprints.main.routes import main_bp
|
||||
from .blueprints.public.routes import public_bp
|
||||
|
||||
from flask import g
|
||||
from flask_login import current_user
|
||||
from .utils.tokens import ensure_fresh_appwrite_jwt
|
||||
# from flask import g
|
||||
# from flask_login import current_user
|
||||
# from .utils.tokens import ensure_fresh_appwrite_jwt
|
||||
|
||||
# load_dotenv()
|
||||
from .utils.settings import get_settings
|
||||
from .utils.session_user import SessionUser
|
||||
|
||||
settings = get_settings()
|
||||
|
||||
@@ -21,41 +22,85 @@ def create_app():
|
||||
APPWRITE_ENDPOINT=settings.appwrite_endpoint,
|
||||
APPWRITE_PROJECT_ID=settings.appwrite_project_id,
|
||||
APPWRITE_API_KEY=settings.appwrite_api_key,
|
||||
SESSION_COOKIE_SECURE = False,
|
||||
SESSION_COOKIE_SAMESITE = "Lax",
|
||||
REMEMBER_COOKIE_SAMESITE = "Lax",
|
||||
REMEMBER_COOKIE_SECURE = False
|
||||
)
|
||||
|
||||
if not app.config["APPWRITE_ENDPOINT"] or not app.config["APPWRITE_PROJECT_ID"]:
|
||||
raise RuntimeError("Missing APPWRITE_ENDPOINT or APPWRITE_PROJECT_ID")
|
||||
|
||||
# Extensions
|
||||
login_manager.init_app(app)
|
||||
login_manager.login_view = "auth.login"
|
||||
|
||||
# Blueprints
|
||||
app.register_blueprint(auth_bp)
|
||||
app.register_blueprint(main_bp)
|
||||
app.register_blueprint(public_bp)
|
||||
|
||||
@app.before_request
|
||||
def _refresh_jwt_if_needed():
|
||||
# Only when logged in; ignore static files etc.
|
||||
if getattr(current_user, "is_authenticated", False):
|
||||
try:
|
||||
# mint if near expiry; otherwise no-op
|
||||
g.appwrite_jwt = ensure_fresh_appwrite_jwt()
|
||||
except Exception:
|
||||
# If the Appwrite session is gone, we don't crash the page;
|
||||
# your protected routes will redirect to login as usual.
|
||||
pass
|
||||
def require_login():
|
||||
"""Gate all routes behind a session 'user' except auth + static."""
|
||||
# Always allow static files
|
||||
if request.endpoint == "static":
|
||||
return
|
||||
|
||||
# Endpoints that should be accessible without being logged in
|
||||
public_endpoints = [
|
||||
"auth.login",
|
||||
"auth.register",
|
||||
"auth.verify",
|
||||
"auth.callback",
|
||||
"auth.send_verification",
|
||||
# add any health checks or webhooks here
|
||||
"public.home",
|
||||
]
|
||||
|
||||
# Make session user easy to access in views/templates
|
||||
g.user = session.get("user")
|
||||
|
||||
endpoint = (request.endpoint or "")
|
||||
|
||||
# Let any route under the auth blueprint through (login/verify/etc.)
|
||||
if endpoint.startswith("public.") or endpoint.startswith("auth."):
|
||||
return
|
||||
|
||||
if endpoint in public_endpoints:
|
||||
return
|
||||
|
||||
|
||||
# Block everything else unless logged in
|
||||
if g.user is None:
|
||||
# preserve destination for GETs
|
||||
next_url = request.url if request.method == "GET" else url_for("auth.login")
|
||||
flash("Please log in to continue.", "warning")
|
||||
return redirect(url_for("auth.login", next=next_url))
|
||||
|
||||
@app.before_request
|
||||
def load_user():
|
||||
user_data = session.get("user")
|
||||
print(user_data)
|
||||
|
||||
if user_data:
|
||||
g.current_user = SessionUser(
|
||||
id=user_data.get("$id",""),
|
||||
registered_on=user_data.get("registration",""),
|
||||
email=user_data.get("email",""),
|
||||
email_verified=user_data.get("emailVerification", False),
|
||||
phone=user_data.get("phone",""),
|
||||
phone_verified=user_data.get("phoneVerification",False),
|
||||
mfa=user_data.get("mfa","")
|
||||
)
|
||||
else:
|
||||
# Anonymous user object with same interface
|
||||
class AnonymousUser:
|
||||
is_authenticated = False
|
||||
email_verification = False
|
||||
|
||||
g.current_user = AnonymousUser()
|
||||
|
||||
|
||||
@app.context_processor
|
||||
def inject_globals():
|
||||
"""Add variables available to all Jinja templates."""
|
||||
return dict(
|
||||
app_name=settings.app_name,
|
||||
app_version=settings.app_version
|
||||
app_version=settings.app_version,
|
||||
current_user=getattr(g, "current_user", None),
|
||||
)
|
||||
|
||||
return app
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
# app/auth/routes.py
|
||||
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
|
||||
from flask import Blueprint, render_template, request, redirect, url_for, flash, session,make_response
|
||||
from flask_login import login_user, logout_user, login_required
|
||||
|
||||
from app.utils.extensions import User, get_client_from_session
|
||||
from app.services.appwrite_client import AppwriteAccountClient
|
||||
from app.services.appwrite_client import AppWriteClient
|
||||
|
||||
|
||||
auth_bp = Blueprint("auth", __name__, url_prefix="/auth")
|
||||
|
||||
auth_bp = Blueprint("auth", __name__, url_prefix="")
|
||||
|
||||
@auth_bp.route("/register", methods=["GET", "POST"])
|
||||
def register():
|
||||
@@ -19,36 +20,21 @@ def register():
|
||||
return redirect(url_for("auth.register"))
|
||||
|
||||
try:
|
||||
# Create user (no bound session needed)
|
||||
aw = AppwriteAccountClient(use_admin=True)
|
||||
aw.create_user(email=email, password=password, name=name)
|
||||
aw = AppWriteClient()
|
||||
aw.create_new_user(email,password,name)
|
||||
|
||||
# Create a session and STORE THE SECRET in Flask session
|
||||
sess_obj = aw.create_email_password_session(email=email, password=password)
|
||||
secret = sess_obj.get("secret")
|
||||
|
||||
session["appwrite_cookies"] = secret
|
||||
|
||||
# Bind and fetch profile (either reuse aw with binding or make a new instance)
|
||||
aw = AppwriteAccountClient(cookies=secret)
|
||||
profile = aw.get_account()
|
||||
session["user_profile"] = profile
|
||||
|
||||
# (Optional) Create a JWT for short-lived API calls
|
||||
jwt = aw.create_jwt()
|
||||
session["appwrite_jwt"] = jwt
|
||||
|
||||
login_user(User(id=profile["$id"], email=profile["email"], name=profile.get("name")))
|
||||
login_valid, error = aw.log_user_in(email,password)
|
||||
if login_valid:
|
||||
flash("Account created and you are now logged in.", "success")
|
||||
return redirect(url_for("main.dashboard"))
|
||||
|
||||
else:
|
||||
flash(str(error), "error")
|
||||
except Exception as e:
|
||||
flash(str(e), "error")
|
||||
return redirect(url_for("auth.register"))
|
||||
|
||||
return render_template("auth/register.html")
|
||||
|
||||
|
||||
@auth_bp.route("/login", methods=["GET", "POST"])
|
||||
def login():
|
||||
if request.method == "POST":
|
||||
@@ -58,65 +44,40 @@ def login():
|
||||
flash("Email and password are required.", "error")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
aw = AppWriteClient()
|
||||
login_valid, error = aw.log_user_in(email,password)
|
||||
try:
|
||||
# 1) Create session once
|
||||
aw = AppwriteAccountClient(use_admin=True)
|
||||
sess_obj = aw.create_email_password_session(email=email, password=password)
|
||||
secret = sess_obj.get("secret")
|
||||
|
||||
# 2) Save the secret for SSR calls
|
||||
session["appwrite_cookies"] = secret
|
||||
|
||||
# 3) Bind and load profile
|
||||
aw = AppwriteAccountClient(cookies=secret)
|
||||
profile = aw.get_account()
|
||||
session["user_profile"] = profile
|
||||
|
||||
# optional: create a JWT now if your dashboard expects it
|
||||
try:
|
||||
session["appwrite_jwt"] = aw.create_jwt()
|
||||
except Exception:
|
||||
session["appwrite_jwt"] = {}
|
||||
|
||||
usersname = profile.get("name")
|
||||
|
||||
login_user(User(id=profile["$id"], email=profile["email"], name=profile.get("name")))
|
||||
flash(f"Welcome back {usersname}!", "success")
|
||||
if login_valid:
|
||||
username = session.get("user",{}).get("name","User")
|
||||
flash(f"Welcome Back {username}", "success")
|
||||
return redirect(url_for("main.dashboard"))
|
||||
else:
|
||||
flash(str(error), "error")
|
||||
except Exception as e:
|
||||
flash(str(e), "error")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
# Get method
|
||||
|
||||
return render_template("auth/login.html")
|
||||
|
||||
|
||||
@auth_bp.route("/logout", methods=["GET", "POST"])
|
||||
@login_required
|
||||
def logout():
|
||||
secret = session.get("appwrite_cookies")
|
||||
if secret:
|
||||
try:
|
||||
aw = AppwriteAccountClient(cookies=secret)
|
||||
aw.logout_current() # best effort
|
||||
except Exception:
|
||||
pass
|
||||
aw = AppWriteClient()
|
||||
aw.log_user_out()
|
||||
session.clear()
|
||||
logout_user()
|
||||
flash("Signed out.", "success")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
@auth_bp.route("/send", methods=["POST"])
|
||||
@login_required
|
||||
def send():
|
||||
"""
|
||||
Sends a verification email to the currently logged-in user.
|
||||
Appwrite will redirect the user back to /verify/callback with userId & secret.
|
||||
"""
|
||||
try:
|
||||
aw = get_client_from_session()
|
||||
callback = url_for("auth.callback", _external=True) # Must be allowed in Appwrite -> Platforms
|
||||
aw.send_verification(callback_url=callback)
|
||||
aw = AppWriteClient()
|
||||
aw.send_email_verification()
|
||||
flash("Verification email sent. Please check your inbox.", "info")
|
||||
except Exception as e:
|
||||
flash(f"Could not send verification email: {e}", "error")
|
||||
@@ -130,27 +91,21 @@ def callback():
|
||||
Requires the user to be logged in (Appwrite Account endpoints need a session).
|
||||
If not logged in, we stash the link params and send them to log in first.
|
||||
"""
|
||||
aw = AppWriteClient()
|
||||
user_id = request.args.get("userId")
|
||||
secret = request.args.get("secret")
|
||||
if not user_id or not secret:
|
||||
flash("Invalid verification link.", "error")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
try:
|
||||
# If we don't currently have an Appwrite session, ask them to log in, then resume.
|
||||
if not session.get("appwrite_cookies"):
|
||||
if not aw.verify_email(user_id,secret):
|
||||
session["pending_verification"] = {"userId": user_id, "secret": secret}
|
||||
flash("Please log in to complete email verification.", "warning")
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
# We have a session; complete verification
|
||||
try:
|
||||
aw = get_client_from_session()
|
||||
aw.complete_verification(user_id=user_id, secret=secret)
|
||||
# Refresh cached profile so templates reflect emailVerification=True
|
||||
profile = aw.get_account()
|
||||
session["user_profile"] = profile
|
||||
# Cleanup any pending state
|
||||
session.pop("pending_verification", None)
|
||||
flash("Email verified! You're all set.", "success")
|
||||
return redirect(url_for("main.dashboard"))
|
||||
except Exception as e:
|
||||
|
||||
@@ -1,18 +1,10 @@
|
||||
from flask import Blueprint, redirect, url_for, render_template, session,flash
|
||||
from flask_login import login_required, current_user
|
||||
from app.services.appwrite_client import AppWriteClient
|
||||
|
||||
main_bp = Blueprint("main", __name__, url_prefix="")
|
||||
aw = AppWriteClient()
|
||||
main_bp = Blueprint("main", __name__, url_prefix="/main")
|
||||
|
||||
@main_bp.route("/")
|
||||
def home():
|
||||
if current_user.is_authenticated:
|
||||
return redirect(url_for("main.dashboard"))
|
||||
return redirect(url_for("auth.login"))
|
||||
|
||||
@main_bp.route("/dashboard")
|
||||
@login_required
|
||||
def dashboard():
|
||||
|
||||
jwt_info = session.get("appwrite_jwt", {})
|
||||
profile = session.get("user_profile", {})
|
||||
return render_template("main/dashboard.html", profile=profile, jwt_info=jwt_info)
|
||||
return render_template("main/dashboard.html", profile="", jwt_info="")
|
||||
|
||||
10
app/blueprints/public/routes.py
Normal file
10
app/blueprints/public/routes.py
Normal file
@@ -0,0 +1,10 @@
|
||||
from flask import Blueprint, redirect, url_for, render_template, session,flash
|
||||
from app.services.appwrite_client import AppWriteClient
|
||||
|
||||
|
||||
public_bp = Blueprint("public", __name__, url_prefix="/")
|
||||
|
||||
|
||||
@public_bp.route("/")
|
||||
def home():
|
||||
return render_template("public/home.html")
|
||||
@@ -1,84 +1,123 @@
|
||||
# app/services/appwrite_client.py
|
||||
from __future__ import annotations
|
||||
import os
|
||||
from typing import Optional, Dict, Any, Mapping, Union
|
||||
from typing import Optional, Dict, Any, Mapping, Union, List
|
||||
|
||||
from flask import current_app, has_request_context, request, session
|
||||
from flask import session, redirect, url_for
|
||||
from appwrite.client import Client
|
||||
from appwrite.services.account import Account
|
||||
from appwrite.id import ID
|
||||
|
||||
|
||||
ENDPOINT = os.getenv("APPWRITE_ENDPOINT")
|
||||
PROJECT_ID = os.getenv("APPWRITE_PROJECT_ID")
|
||||
API_KEY = os.getenv("APPWRITE_API_KEY")
|
||||
|
||||
class AppwriteAccountClient:
|
||||
def __init__(self, cookies: Optional[Union[str, Mapping[str, str]]] = None, use_admin: bool = False) -> None:
|
||||
endpoint = current_app.config.get("APPWRITE_ENDPOINT") or os.getenv("APPWRITE_ENDPOINT")
|
||||
project_id = current_app.config.get("APPWRITE_PROJECT_ID") or os.getenv("APPWRITE_PROJECT_ID")
|
||||
api_key = current_app.config.get("APPWRITE_API_KEY") or os.getenv("APPWRITE_API_KEY")
|
||||
if not endpoint or not project_id:
|
||||
raise RuntimeError("APPWRITE_ENDPOINT and APPWRITE_PROJECT_ID must be configured")
|
||||
# SESSION USER OBJECT DICT NOTES
|
||||
# {
|
||||
# "$id": "6902663c000efa514a81",
|
||||
# "$createdAt": "2025-10-29T19:08:44.483+00:00",
|
||||
# "$updatedAt": "2025-10-31T00:28:26.422+00:00",
|
||||
# "name": "Test Account",
|
||||
# "registration": "2025-10-29T19:08:44.482+00:00",
|
||||
# "status": true,
|
||||
# "labels": [],
|
||||
# "passwordUpdate": "2025-10-29T19:08:44.482+00:00",
|
||||
# "email": "ptarrant@gmail.com",
|
||||
# "phone": "",
|
||||
# "emailVerification": false,
|
||||
# "phoneVerification": false,
|
||||
# "mfa": false,
|
||||
# "prefs": {},
|
||||
# "targets": [
|
||||
# {
|
||||
# "$id": "6902663c81f9f1a63f4c",
|
||||
# "$createdAt": "2025-10-29T19:08:44.532+00:00",
|
||||
# "$updatedAt": "2025-10-29T19:08:44.532+00:00",
|
||||
# "name": "",
|
||||
# "userId": "6902663c000efa514a81",
|
||||
# "providerId": null,
|
||||
# "providerType": "email",
|
||||
# "identifier": "ptarrant@gmail.com",
|
||||
# "expired": false
|
||||
# }
|
||||
# ],
|
||||
# "accessedAt": "2025-10-31T00:28:26.418+00:00"
|
||||
# }
|
||||
|
||||
self.endpoint = endpoint
|
||||
self.project_id = project_id
|
||||
class AppWriteClient:
|
||||
def __init__(self):
|
||||
self.session_key = f"a_session_{PROJECT_ID}"
|
||||
|
||||
self.client = Client()
|
||||
self.client.set_endpoint(self.endpoint)
|
||||
self.client.set_project(self.project_id)
|
||||
def _get_admin_client(self):
|
||||
return (Client()
|
||||
.set_endpoint(ENDPOINT)
|
||||
.set_project(PROJECT_ID)
|
||||
.set_key(API_KEY)
|
||||
)
|
||||
|
||||
# If we need admin privileges (to get session.secret), set the API key
|
||||
if use_admin:
|
||||
if not api_key:
|
||||
raise RuntimeError("APPWRITE_API_KEY is required when use_admin=True")
|
||||
self.client.set_key(api_key)
|
||||
def _get_user_client(self):
|
||||
client = (Client()
|
||||
.set_endpoint(ENDPOINT)
|
||||
.set_project(PROJECT_ID)
|
||||
)
|
||||
|
||||
# Bind session if available (explicit → browser cookie → Flask session)
|
||||
bound = False
|
||||
if cookies:
|
||||
bound = self._bind_session_from(cookies)
|
||||
if not bound and has_request_context():
|
||||
bound = self._bind_session_from(request.cookies)
|
||||
if not bound and has_request_context():
|
||||
secret = session.get("appwrite_cookies")
|
||||
if secret:
|
||||
self.client.set_session(secret)
|
||||
bound = True
|
||||
if session[self.session_key] is not None:
|
||||
client.set_session(session[self.session_key])
|
||||
|
||||
self.account = Account(self.client)
|
||||
return client
|
||||
|
||||
@staticmethod
|
||||
def session_cookie_key(project_id: str) -> str:
|
||||
return f"a_session_{project_id}"
|
||||
def create_new_user(self, email:str, password:str, name:Optional[str]):
|
||||
admin_client = self._get_admin_client()
|
||||
try:
|
||||
admin_account = Account(admin_client)
|
||||
admin_account.create(user_id=ID.unique(),email=email,password=password,name=name)
|
||||
return True, ""
|
||||
except Exception as e:
|
||||
return False, e
|
||||
|
||||
def _bind_session_from(self, cookies: Union[str, Mapping[str, str]]) -> bool:
|
||||
if isinstance(cookies, str):
|
||||
self.client.set_session(cookies); return True
|
||||
key = f"a_session_{self.project_id}"
|
||||
if key in cookies and cookies.get(key):
|
||||
self.client.set_session(cookies[key]); return True
|
||||
for v in cookies.values():
|
||||
if v: self.client.set_session(v); return True
|
||||
return False
|
||||
def _refresh_user_session_data(self):
|
||||
user_client = self._get_user_client()
|
||||
user_account = Account(user_client)
|
||||
user = user_account.get()
|
||||
session['user']=user
|
||||
|
||||
# --- Auth & account helpers ---
|
||||
def create_user(self, email: str, password: str, name: Optional[str] = None, user_id: Optional[str] = None) -> Dict[str, Any]:
|
||||
return dict(self.account.create(user_id=user_id or ID.unique(), email=email, password=password, name=name))
|
||||
def log_user_in(self, email:str,password:str):
|
||||
admin_client = self._get_admin_client()
|
||||
try:
|
||||
admin_account = Account(admin_client)
|
||||
user_session = admin_account.create_email_password_session(email,password)
|
||||
session[self.session_key]=user_session['secret']
|
||||
|
||||
def create_email_password_session(self, email: str, password: str) -> Dict[str, Any]:
|
||||
return dict(self.account.create_email_password_session(email=email, password=password))
|
||||
self._refresh_user_session_data()
|
||||
|
||||
def create_jwt(self) -> Dict[str, Any]:
|
||||
return dict(self.account.create_jwt())
|
||||
return True, ""
|
||||
except Exception as e:
|
||||
return False, str(e)
|
||||
|
||||
def get_account(self) -> Dict[str, Any]:
|
||||
return dict(self.account.get())
|
||||
|
||||
def logout_current(self) -> bool:
|
||||
self.account.delete_session("current")
|
||||
def log_user_out(self):
|
||||
try:
|
||||
user_client = self._get_user_client()
|
||||
user_account = Account(user_client)
|
||||
user_account.delete_sessions()
|
||||
return True
|
||||
except Exception as e:
|
||||
return True
|
||||
|
||||
# --- Email verification ---
|
||||
def send_verification(self, callback_url: str) -> Dict[str, Any]:
|
||||
return dict(self.account.create_verification(url=callback_url))
|
||||
def send_email_verification(self):
|
||||
user_client = self._get_user_client()
|
||||
user_account = Account(user_client)
|
||||
callback_url = url_for('auth.callback', _external=True)
|
||||
user_account.create_verification(url=callback_url)
|
||||
|
||||
def complete_verification(self, user_id: str, secret: str) -> Dict[str, Any]:
|
||||
return dict(self.account.update_verification(user_id=user_id, secret=secret))
|
||||
def verify_email(self, user_id:str, secret:str):
|
||||
if session[self.session_key] is None:
|
||||
return False
|
||||
try:
|
||||
user_client = self._get_user_client()
|
||||
user_account = Account(user_client)
|
||||
user_account.update_email_verification(user_id,secret)
|
||||
self._refresh_user_session_data()
|
||||
return True
|
||||
except Exception as e:
|
||||
return False
|
||||
@@ -46,6 +46,7 @@
|
||||
</div>
|
||||
{% endif %}
|
||||
|
||||
|
||||
{% block body %}
|
||||
<!-- Page content goes here -->
|
||||
{% endblock %}
|
||||
|
||||
33
app/templates/public/home.html
Normal file
33
app/templates/public/home.html
Normal file
@@ -0,0 +1,33 @@
|
||||
{% extends "bases/login_base.html" %}
|
||||
{% block content %}
|
||||
<div class="card shadow-lg p-4" style="max-width: 700px; width: 100%;text-align: justify;">
|
||||
<div class="card-body text-center">
|
||||
<h5 class="card-title mb-4">
|
||||
<i class="fa-solid fa-right-to-bracket me-2"></i> Welcome to Code of Conquest
|
||||
</h5>
|
||||
|
||||
<p style="text-align: justify;">
|
||||
In the world of Code of Conquest, the line between hero and villain blurs as you embark on a legendary adventure.
|
||||
This immersive game drops you into a realm of wonder and danger, where every decision, every action, and every
|
||||
roll of the dice determines the fate of your character.<br /><br />
|
||||
|
||||
With our revolutionary AI-driven DM, each playthrough is unique, offering a fresh challenge tailored to your
|
||||
choices. Explore ancient ruins, mysterious forests, and forgotten cities, teeming with hidden secrets, fearsome
|
||||
monsters, and legendary treasures.<br /><br />
|
||||
|
||||
Master the art of combat, magic, and stealth as you navigate a world of intrigue and deception. Forge alliances,
|
||||
rivalries, and friendships that will shape the course of history. Will you rise to become a legendary conqueror or
|
||||
succumb to the darkness within?<br /><br />
|
||||
|
||||
Code of Conquest is not just a game – it's an experience that lets you forge your own legend. With every triumph,
|
||||
you'll earn reputation, wealth, and power. Your name will echo through the annals of history as you make choices
|
||||
that shape the world around you.
|
||||
</p>
|
||||
|
||||
</div>
|
||||
<div class="card-footer text-center">
|
||||
<small>No account?</small>
|
||||
<a href="{{ url_for('auth.register') }}" class="btn btn-link p-0 ms-1">Create one</a>
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -1,56 +0,0 @@
|
||||
# app/utils/extensions.py
|
||||
from flask_login import LoginManager
|
||||
from flask import session
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
from app.services.appwrite_client import AppwriteAccountClient
|
||||
|
||||
login_manager = LoginManager()
|
||||
|
||||
@dataclass
|
||||
class User:
|
||||
id: str
|
||||
email: str
|
||||
name: Optional[str] = None
|
||||
email_verification: bool = False
|
||||
|
||||
def is_active(self): return True
|
||||
def is_authenticated(self): return True
|
||||
def is_anonymous(self): return False
|
||||
def get_id(self): return self.id
|
||||
|
||||
@login_manager.user_loader
|
||||
def load_user(user_id: str) -> Optional[User]:
|
||||
# First: use cached profile
|
||||
u = session.get("user_profile")
|
||||
if u and u.get("$id") == user_id:
|
||||
return User(
|
||||
id=u["$id"],
|
||||
email=u["email"],
|
||||
name=u.get("name"),
|
||||
email_verification=bool(u.get("emailVerification", False)),
|
||||
)
|
||||
|
||||
# Next: use the session secret we stored at login
|
||||
secret = session.get("appwrite_cookies")
|
||||
if not secret:
|
||||
return None
|
||||
|
||||
aw = AppwriteAccountClient(cookies=secret)
|
||||
try:
|
||||
acc = aw.get_account()
|
||||
session["user_profile"] = acc
|
||||
return User(
|
||||
id=acc["$id"],
|
||||
email=acc["email"],
|
||||
name=acc.get("name"),
|
||||
email_verification=bool(acc.get("emailVerification", False)),
|
||||
)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def get_client_from_session() -> AppwriteAccountClient:
|
||||
secret = session.get("appwrite_cookies")
|
||||
if not secret:
|
||||
raise RuntimeError("No Appwrite session is available. Please log in.")
|
||||
return AppwriteAccountClient(cookies=secret)
|
||||
27
app/utils/session_user.py
Normal file
27
app/utils/session_user.py
Normal file
@@ -0,0 +1,27 @@
|
||||
from dataclasses import dataclass
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class SessionUser:
|
||||
id: str = ""
|
||||
registered_on: str = ""
|
||||
name: str = ""
|
||||
email: str = ""
|
||||
email_verified: bool = False
|
||||
phone: str = ""
|
||||
phone_verified: bool = False
|
||||
mfa: bool = False
|
||||
|
||||
|
||||
@property
|
||||
def is_authenticated(self) -> bool:
|
||||
return True
|
||||
|
||||
@property
|
||||
def email_verification(self) -> bool:
|
||||
return self.email_verified
|
||||
|
||||
@property
|
||||
def phone_verification(self) -> bool:
|
||||
return self.phone_verification
|
||||
Reference in New Issue
Block a user