Platform /me Endpoint Bugfix

Database lookup fix for /auth/me endpoint to return complete user data with created_at

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

Quick Start

1. 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": "[e-mail]",
               "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())
   
2. 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)

@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
    )

After (Fixed)

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

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": "[e-mail]",
            "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"] == "[e-mail]"

Integration Test

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": "[e-mail]",
            "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"

Error Case 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="[e-mail]")
            
            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

---

Related

Platform Authentication

Complete authentication setup and usage

Platform API Reference

Full platform API documentation