49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
import secrets
|
|
from database import db
|
|
from models import Session, User, Organization
|
|
from flask import jsonify
|
|
|
|
|
|
class SessionService:
|
|
@staticmethod
|
|
def create_session(user: User, org: Organization) -> Session:
|
|
session = Session(
|
|
user_id=user.id,
|
|
org_id=org.id,
|
|
token=secrets.token_hex(128)
|
|
)
|
|
db.add(session)
|
|
db.commit()
|
|
db.refresh(session)
|
|
return session
|
|
|
|
@staticmethod
|
|
def get_session(token: str) -> Session | None:
|
|
return db.query(Session).filter(Session.token == token).first()
|
|
|
|
@staticmethod
|
|
def delete_session(session: Session) -> None:
|
|
db.delete(session)
|
|
db.commit()
|
|
|
|
@staticmethod
|
|
def validate_session(token: str) -> tuple | Session:
|
|
from services import OrganizationService
|
|
|
|
if "Bearer" in token:
|
|
token = token.split(" ")[1]
|
|
|
|
session = SessionService.get_session(token)
|
|
if not session:
|
|
return jsonify({"error": "Not authenticated"}), 401
|
|
|
|
org = OrganizationService.get_organization(session.org_id)
|
|
if not org:
|
|
return jsonify({"error": "Organization not found"}), 404
|
|
|
|
status = OrganizationService.get_user_status(org, session.user_id)
|
|
if status != "active":
|
|
return jsonify({"error": "User is not active"}), 403
|
|
|
|
return session
|