Platform /me Endpoint Bugfix

Platform Authentication /me endpoint now returns complete user data by querying the database instead of using incomplete token data.

Quick Start

Test Fixed Endpoint

The /auth/me endpoint now returns complete user data including the created_at timestamp:

import httpx
import asyncio

async def test_fixed_endpoint():
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        # Login to get token
        auth_response = await client.post("/api/v1/auth/login", json={
            "email": "user@example.com",
            "password": "password"
        })
        
        token = auth_response.json()["token"]
        
        # Test the fixed /me endpoint
        headers = {"Authorization": f"Bearer {token}"}
        me_response = await client.get("/api/v1/auth/me", headers=headers)
        user_data = me_response.json()
        
        # Verify created_at is now present
        assert user_data["created_at"] is not None
        print(f"✅ created_at: {user_data['created_at']}")
        print(f"✅ Full user data: {user_data}")

asyncio.run(test_fixed_endpoint())

Verify Error Handling

The endpoint now also returns 404 if the user token is valid but the user record is missing from the database:

import httpx
import asyncio

async def test_error_handling():
    # This would happen if a user token is valid but user was deleted
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        headers = {"Authorization": "Bearer VALID_TOKEN_BUT_USER_DELETED"}
        
        try:
            response = await client.get("/api/v1/auth/me", headers=headers)
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                print("✅ Correctly returns 404 when user not found")
                print(f"Error: {e.response.json()['detail']}")

asyncio.run(test_error_handling())

How It Works

Issue	Before Fix	After Fix
Data Source	JWT token only	Database query
created_at	Always `None`	Actual timestamp
Completeness	Incomplete data	Full user record
Error Handling	No validation	404 if user not found

Technical Implementation

Code Changes

The fix involved modifying the /auth/me endpoint in src/praisonai-platform/praisonai_platform/api/routes/auth.py:

Before (Broken)
After (Fixed)

@router.get("/me", response_model=UserResponse)
async def me(current_user: AuthIdentity = Depends(get_current_user)):
    return UserResponse(
        id=current_user.id,
        name=current_user.name or "",
        email=current_user.email or "",
        created_at=None,  # ❌ Always None - data not in JWT
    )

from sqlalchemy import select
from ...models import User

@router.get("/me", response_model=UserResponse)
async def me(
    current_user: AuthIdentity = Depends(get_current_user),
    session: AsyncSession = Depends(get_db)
):
    # Query database for complete user record
    result = await session.execute(select(User).where(User.id == current_user.id))
    db_user = result.scalar_one_or_none()
    
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    
    return UserResponse(
        id=db_user.id,
        email=db_user.email,
        name=db_user.name,
        created_at=db_user.created_at  # ✅ Real timestamp from database
    )

Root Cause Analysis

Component	Issue	Resolution
JWT Token	Contains minimal user data (id, email, name)	Added database lookup for complete data
AuthIdentity	No `created_at` field in token payload	Query User model from database
Validation	No check if user still exists	Added 404 error for missing users
Performance	N/A	Single DB query per request (acceptable for /me)

Testing the Fix

Verify the bugfix with these test scenarios:

Unit Test
Integration Test
Error Case Test

import pytest
from fastapi.testclient import TestClient
from praisonai_platform.api.app import create_app

@pytest.mark.asyncio
async def test_me_endpoint_returns_created_at():
    """Test that /me endpoint returns complete user data with created_at."""
    app = create_app()
    
    with TestClient(app) as client:
        # Register user
        register_response = client.post("/api/v1/auth/register", json={
            "email": "test@example.com",
            "password": "testpass123",
            "name": "Test User"
        })
        assert register_response.status_code == 201
        
        token = register_response.json()["token"]
        headers = {"Authorization": f"Bearer {token}"}
        
        # Test /me endpoint
        me_response = client.get("/api/v1/auth/me", headers=headers)
        assert me_response.status_code == 200
        
        user_data = me_response.json()
        assert user_data["created_at"] is not None
        assert "T" in user_data["created_at"]  # ISO format timestamp
        assert user_data["name"] == "Test User"
        assert user_data["email"] == "test@example.com"

import httpx
import asyncio
import pytest

@pytest.mark.asyncio
async def test_me_endpoint_integration():
    """Integration test for the fixed /me endpoint."""
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        # Create test user
        register_data = {
            "email": "integration@test.com",
            "password": "secure123",
            "name": "Integration Test"
        }
        
        register_resp = await client.post("/api/v1/auth/register", json=register_data)
        assert register_resp.status_code == 201
        
        token = register_resp.json()["token"]
        created_at_from_register = register_resp.json()["user"]["created_at"]
        
        # Test /me endpoint 
        headers = {"Authorization": f"Bearer {token}"}
        me_resp = await client.get("/api/v1/auth/me", headers=headers)
        assert me_resp.status_code == 200
        
        me_data = me_resp.json()
        
        # Verify fix: created_at should match registration
        assert me_data["created_at"] == created_at_from_register
        assert me_data["created_at"] is not None
        assert me_data["name"] == "Integration Test"

import pytest
from unittest.mock import AsyncMock, patch
from fastapi.testclient import TestClient
from praisonai_platform.api.app import create_app

@pytest.mark.asyncio  
async def test_me_endpoint_user_not_found():
    """Test /me endpoint returns 404 when user record is missing."""
    app = create_app()
    
    with TestClient(app) as client:
        # Mock scenario: valid token but user deleted from DB
        with patch('praisonai_platform.api.routes.auth.get_current_user') as mock_auth:
            # Mock AuthIdentity with valid user data
            mock_auth.return_value = AsyncMock(id="deleted_user_123", email="test@example.com")
            
            with patch('sqlalchemy.ext.asyncio.AsyncSession.execute') as mock_execute:
                # Mock database returning no user
                mock_result = AsyncMock()
                mock_result.scalar_one_or_none.return_value = None
                mock_execute.return_value = mock_result
                
                headers = {"Authorization": "Bearer valid_token"}
                response = client.get("/api/v1/auth/me", headers=headers)
                
                assert response.status_code == 404
                assert "User not found" in response.json()["detail"]

Common Patterns

Database Query Optimization

The fix adds one database query per /me request. For high-traffic applications, consider caching:

from functools import lru_cache
import asyncio

@lru_cache(maxsize=1000)
async def get_cached_user(user_id: str, ttl: int = 300):
    """Cache user data for 5 minutes to reduce DB load."""
    # Implementation would include cache invalidation
    # This is a simplified example
    pass

@router.get("/me", response_model=UserResponse)
async def me(current_user: AuthIdentity = Depends(get_current_user)):
    # Try cache first, fallback to database
    cached_user = await get_cached_user(current_user.id)
    if cached_user:
        return UserResponse.model_validate(cached_user)
    
    # Fallback to database query...

JWT Token Enhancement

To avoid database queries entirely, include created_at in JWT tokens for new registrations:

def create_jwt_token(user: User) -> str:
    """Create JWT token with complete user data."""
    payload = {
        "sub": user.id,
        "email": user.email, 
        "name": user.name,
        "created_at": user.created_at.isoformat(),  # Include in token
        "exp": datetime.utcnow() + timedelta(seconds=JWT_TTL)
    }
    return jwt.encode(payload, JWT_SECRET, algorithm="HS256")

Note: This requires token regeneration for existing users.

Monitoring and Alerting

Monitor the /me endpoint for 404 errors which indicate data consistency issues:

import logging
from prometheus_client import Counter

# Metrics
me_endpoint_404_counter = Counter('auth_me_404_total', 'User not found errors')

@router.get("/me", response_model=UserResponse)
async def me(current_user: AuthIdentity = Depends(get_current_user), session: AsyncSession = Depends(get_db)):
    result = await session.execute(select(User).where(User.id == current_user.id))
    db_user = result.scalar_one_or_none()
    
    if db_user is None:
        # Log and track data consistency issues
        logging.error(f"Valid JWT token but user {current_user.id} not found in database")
        me_endpoint_404_counter.inc()
        raise HTTPException(status_code=404, detail="User not found")
    
    return UserResponse.model_validate(db_user)

Best Practices

Data Consistency

JWT vs Database: Keep token data minimal to avoid sync issues
Database as Source: Always query database for complete, up-to-date data
Error Handling: Return 404 when token is valid but user record is missing
Monitoring: Track 404s from /me endpoint to detect data consistency issues

Performance Considerations

Single Query: One DB query per /me request is acceptable for most use cases
Caching Strategy: Consider Redis caching for high-traffic scenarios
Connection Pooling: Ensure database connection pooling is configured
Query Optimization: Use database indexes on user.id for fast lookups

Security Improvements

Token Validation: Verify JWT signature before database lookup
Rate Limiting: Apply rate limiting to prevent abuse of /me endpoint
Audit Logging: Log 404 cases for security investigation
Input Validation: Validate user.id format from JWT payload

Testing Strategy

Unit Tests: Test both success and error cases
Integration Tests: Verify complete request/response flow
Performance Tests: Ensure acceptable response times under load
Security Tests: Test with malformed or expired tokens

Platform Authentication

Complete authentication setup and usage

Platform API Reference

Full platform API documentation

Getting Started

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

Platform /me Endpoint Bugfix

Quick Start

How It Works

Technical Implementation

Code Changes

Root Cause Analysis

Testing the Fix

Common Patterns

Best Practices

Platform Authentication

Platform API Reference

Getting Started

Core Concepts

Guides

Features

Models

Databases

Observability

Memory

Knowledge

RAG

Persistence

Tools

Other Features

Developers

Configuration

Best Practices

Getting Started (No Code)

​Quick Start

​How It Works

​Technical Implementation

​Code Changes

​Root Cause Analysis

​Testing the Fix

​Common Patterns

​Best Practices

​Related

Platform Authentication

Platform API Reference

Quick Start

How It Works

Technical Implementation

Code Changes

Root Cause Analysis

Testing the Fix

Common Patterns

Best Practices

Related