feat: implement user session management with jwt and database persistence

api/utils/celery_config.py

@@ -0,0 +1,46 @@
+from celery import Celery
+from celery.schedules import crontab
+from datetime import datetime, timezone
+
+from api.utils.settings import settings
+
+celery_app = Celery('api', broker='redis://localhost:6379/0', backend='redis://localhost:6379/0')
+
+celery_app.conf.beat_schedule = {
+    "clean_db_every_day": {
+        "task": "api.utils.celery_config.clean_expired_and_revoked_tokens_from_sessions_table",
+        "schedule": crontab(hour=0, minute=0) # run everyday at midnight
+    }
+}
+
+celery_app.conf.timezone = "UTC"
+
+# Additional configurations
+celery_app.conf.update(
+    task_serializer='json',
+    result_serializer='json',
+    accept_content=['json'],
+    task_acks_late=True,
+    worker_prefetch_multiplier=1,
+    task_time_limit=300,
+    task_soft_time_limit=180,
+)
+
+@celery_app.task
+def clean_expired_and_revoked_tokens_from_sessions_table():
+    """Clean expired and revoked tokens from the sessions table."""
+    from api.db.database import get_db
+    from api.v1.models.session import UserSession
+    from sqlalchemy import cast, DateTime
+
+    db = next(get_db())
+    try:
+        current_time = datetime.now(timezone.utc)
+        db.query(UserSession).filter(UserSession.is_revoked == True).delete()
+        db.query(UserSession).filter(cast(UserSession.expires_at, DateTime) < current_time).delete()
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        raise e
+    finally:
+        db.close()

api/utils/session_helpers.py

@@ -0,0 +1,73 @@
+import httpx
+from fastapi import Request, BackgroundTasks
+from sqlalchemy.orm import Session
+
+from api.db.database import get_db
+from api.v1.schemas.session import SessionCreate
+from api.v1.services.session import session_service
+from api.utils.client_helpers import get_ip_address
+
+
+async def get_ip_location(ip):
+    """ Get IP location.
+
+    Args:
+        ip (str): IP address
+    """
+    country = region = "Unknown"
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(f"https://ipinfo.io/{ip}/json/", timeout=10)
+            response.raise_for_status()
+            data = response.json()
+            region = data.get("region", "Unknown")
+            country = data.get("country", "Unknown")
+    except httpx.RequestError as exc:
+        print(f"An error occurred while requesting {exc.request.url!r}.")
+    except httpx.HTTPStatusError as exc:
+        print(f"Error response {exc.response.status_code} while requesting {exc.request.url!r}.")
+    except Exception as exc:
+        print(f"An unexpected error occurred: {exc}")
+
+    return f"{region}, {country}"
+
+async def get_session_schema_data(request: Request, refresh_token: str = "", expires_at: str = ""):
+    """Get session schema data.
+
+    Args:
+        request (Request): Request object
+        refresh_token (str): Refresh token
+        expires_at (str): Expiry date
+    """
+    ip = get_ip_address(request)
+    return SessionCreate(
+        ip_address=ip,
+        location=await get_ip_location(ip),
+        device=request.headers.get("User-Agent"),
+        is_revoked=False,
+        refresh_token=refresh_token,
+        expires_at=expires_at
+    )
+
+async def create_session_for_user(
+        db: Session,
+        request: Request,
+        user_id: str,
+        refresh_token: str = "",
+        expires_at: str = "",
+    ):
+    """Create session for user.
+
+    Args:
+        request (Request): Request object
+        db: Database session
+        refresh_token (str): Refresh token
+        expires_at (str): Expiry date
+    """
+    session_data: SessionCreate = await get_session_schema_data(
+        request,
+        refresh_token=refresh_token,
+        expires_at=expires_at,
+    )
+    session_service.create(db, schema=session_data, user_id=user_id)

api/v1/models/__init__.py

@@ -33,4 +33,4 @@
 from api.v1.models.wishlist import Wishlist
 from api.v1.models.totp_device import TOTPDevice
 from api.v1.models.bookmark import Bookmark
-
+from api.v1.models.session import UserSession

api/v1/models/session.py

@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+"""The Session Model."""
+
+from api.v1.models.base_model import BaseTableModel
+from sqlalchemy import Column, String, DateTime, ForeignKey, Boolean, text
+from sqlalchemy.orm import relationship
+
+
+class UserSession(BaseTableModel):
+    __tablename__ = "sessions"
+
+    user_id = Column(String, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
+    ip_address = Column(String, nullable=False)
+    location = Column(String, nullable=True)
+    device = Column(String, nullable=True)
+    is_revoked = Column(Boolean, server_default=text("false"))
+    refresh_token = Column(String, nullable=False)
+    expires_at = Column(DateTime, nullable=False)
+
+    user = relationship("User", back_populates="sessions")
+
+    def __str__(self):
+        return f"{self.user_id} - {self.ip_address}"

api/v1/models/user.py

@@ -34,6 +34,7 @@ class User(BaseTableModel):
     profile = relationship(
         "Profile", uselist=False, back_populates="user", cascade="all, delete-orphan"
     )
+
     organisations = relationship(
         "Organisation", secondary=user_organisation_roles, back_populates="users"
     )
@@ -117,6 +118,9 @@ class User(BaseTableModel):
         "Bookmark", back_populates="user", cascade="delete"
     )
 
+    sessions = relationship(
+        "UserSession", back_populates="user", cascade="all, delete-orphan"
+    )
     def to_dict(self):
         obj_dict = super().to_dict()
         obj_dict.pop("password")

api/v1/routes/__init__.py

@@ -45,6 +45,7 @@
 from api.v1.routes.privacy import privacies
 from api.v1.routes.settings import settings
 from api.v1.routes.terms_and_conditions import terms_and_conditions
+from api.v1.routes.session import session_router
 from api.v1.routes.stripe import subscription_
 from api.v1.routes.wishlist import wishlist
 
@@ -98,3 +99,4 @@
 api_version_one.include_router(product_comment)
 api_version_one.include_router(subscription_)
 api_version_one.include_router(wishlist)
+api_version_one.include_router(session_router)

api/v1/routes/auth.py

@@ -1,4 +1,5 @@
 import logging
+import datetime as dt
 from datetime import timedelta
 from fastapi.responses import JSONResponse
 from jose import ExpiredSignatureError, JWTError
@@ -21,6 +22,8 @@
 from api.core.dependencies.email_sender import send_email
 from api.utils.success_response import auth_response, success_response
 from api.utils.send_mail import send_magic_link
+from api.utils.settings import settings
+from api.utils.session_helpers import create_session_for_user
 from api.v1.models import User
 from api.v1.schemas.user import Token, UserEmailSender
 from api.v1.schemas.user import (
@@ -31,7 +34,6 @@
     UserData2,
 )
 from api.v1.schemas.token import TokenRequest
-
 from api.v1.schemas.user import (MagicLinkRequest,
                                  ChangePasswordSchema,
                                  AuthMeResponse)
@@ -50,6 +52,7 @@
 )
 from api.v1.services.totp import totp_service
 from api.utils.settings import settings
+from api.v1.services.session import session_service
 
 auth = APIRouter(prefix="/auth", tags=["Authentication"])
 
@@ -75,14 +78,9 @@ def register(
     # Create user account
     user = user_service.create(db=db, schema=user_schema)
 
-
     verification_token = user_service.create_verification_token(user.id)
     verification_link = f"{base_url}/api/v1/auth/verify-email?token={verification_token}"
 
-    access_token = user_service.create_access_token(user_id=user.id)
-    refresh_token = user_service.create_refresh_token(user_id=user.id)
-    cta_link = "https://anchor-python.teams.hng.tech/about-us"
-
     # create an organization for the user
     org = CreateUpdateOrganisation(
         name=f"{user.email}'s Organisation", email=user.email
@@ -95,6 +93,18 @@ def register(
     refresh_token = user_service.create_refresh_token(user_id=user.id)
     cta_link = f"{settings.ANCHOR_PYTHON_BASE_URL}/about-us"
 
+    # create session for user
+    expires = dt.datetime.now(dt.timezone.utc) + (dt.timedelta(
+        days=settings.JWT_REFRESH_EXPIRY) - dt.timedelta(seconds=1)
+    )
+    background_tasks.add_task(
+        create_session_for_user,
+        db=db,
+        request=request,
+        user_id=user.id,
+        refresh_token=refresh_token,
+        expires_at=expires
+    )
 
     # Send email in the background
     background_tasks.add_task(
@@ -246,6 +256,19 @@ def login(request: Request, login_request: LoginRequest, background_tasks: Backg
     access_token = user_service.create_access_token(user_id=user.id)
     refresh_token = user_service.create_refresh_token(user_id=user.id)
 
+    # create session for user
+    expires = dt.datetime.now(dt.timezone.utc) + (dt.timedelta(
+        days=settings.JWT_REFRESH_EXPIRY) - dt.timedelta(seconds=1)
+    )
+    background_tasks.add_task(
+        create_session_for_user,
+        db=db,
+        request=request,
+        user_id=user.id,
+        refresh_token=refresh_token,
+        expires_at=expires
+    )
+
     # Background task for email notification
     logger.info(f"Queueing login notification for {user.email} in the background...")
     background_tasks.add_task(send_login_notification, user, request)
@@ -285,6 +308,11 @@ def logout(
 ):
     """Endpoint to log a user out of their account"""
 
+    # logout/delete current user session    
+    current_refresh_token = request.cookies.get("refresh_token")
+    session_service.logout_session(db, current_user.id, current_refresh_token)
+
+
     response = success_response(status_code=200, message="User logged put successfully")
 
     # Delete refresh token from cookies

api/v1/routes/session.py

@@ -0,0 +1,89 @@
+from sqlalchemy.orm import Session
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.encoders import jsonable_encoder
+from fastapi.responses import JSONResponse
+
+from api.v1.models import User
+from api.db.database import get_db
+from api.utils.success_response import success_response
+from api.v1.services.user import user_service
+from api.v1.services.session import session_service
+
+
+session_router = APIRouter(prefix="/sessions", tags=["sessions"])
+
+@session_router.get("/", response_model=success_response)
+def get_all_sessions(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(user_service.get_current_user),
+):
+    """
+    Endpoint to get all sessions.
+
+    args:
+        - db: the database session
+        - current_user: current authenticated user
+    """
+    sessions = session_service.fetch_all(db, current_user.id) 
+    return success_response(
+        status_code=status.HTTP_200_OK,
+        message="Sessions retrieved successfully",
+        data=jsonable_encoder(sessions, exclude={"refresh_token"})
+    )
+
+@session_router.get('/{session_id}', response_model=success_response)
+def get_session(
+    session_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(user_service.get_current_user)
+):
+    session = session_service.fetch(db, current_user.id, session_id)
+    return success_response(
+        status_code=status.HTTP_200_OK,
+        message="Session retrieved successfully",
+        data=jsonable_encoder(session, exclude={"refresh_token"})
+    )
+
+@session_router.delete('/{session_id}', status_code=status.HTTP_204_NO_CONTENT)
+def delete_session(
+    session_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(user_service.get_current_user)
+):
+    """
+    Endpoint to delete a session.
+
+    args:
+        - session_id (str): ID of the session
+        - db: the database session
+        - current_user: current authenticated user
+    """
+    session_service.delete(db, current_user.id, session_id)
+    response_data = {
+        "status": "success",
+        "status_code": 204,
+        "message": "Session deleted successfully",
+        "data": {}
+    }
+    return JSONResponse(
+        status_code=status.HTTP_204_NO_CONTENT,
+        content=jsonable_encoder(response_data)
+    )
+
+@session_router.delete('/', status_code=status.HTTP_204_NO_CONTENT)
+def delete_all_sessions(
+    db: Session = Depends(get_db),
+    current_user: User = Depends(user_service.get_current_user)
+):
+    session_service.delete_all(db, current_user.id)
+    response_data = {
+        "status": "success",
+        "status_code": 204,
+        "message": "Sessions deleted successfully",
+        "data": {}
+    }
+    return JSONResponse(
+        status_code=status.HTTP_204_NO_CONTENT,
+        content=jsonable_encoder(response_data)
+    )

api/v1/schemas/session.py

@@ -0,0 +1,10 @@
+from datetime import datetime
+from pydantic import BaseModel, Field
+
+class SessionCreate(BaseModel):
+    ip_address: str
+    location: str = None
+    device: str = None
+    is_revoked: bool = False
+    refresh_token: str
+    expires_at: datetime