User Manager API

A FastAPI-based user management system with SQLAlchemy and PostgreSQL.

Features

• User Management: Create, read, update, and delete users
• Authentication: User authentication with user_id and token
• WebSocket Support: Real-time communication with token-based auth
• Message Storage: Store and retrieve WebSocket messages
• Quiz System: Real-time quiz games between two users
• Question Management: Create and manage quiz questions
• Async Operations: Full async support with SQLAlchemy 2.0
• Data Validation: Pydantic v2 models for request/response validation
• Database: PostgreSQL with asyncpg driver
• API Documentation: Automatic OpenAPI/Swagger docs

Data Model

Users have the following fields:

• id: Auto-generated primary key
• user_id: Unique user identifier (string, max 255 chars)
• token: Authentication token (string, max 500 chars)

Setup

Prerequisites

• Python 3.9+
• PostgreSQL database
• Virtual environment (recommended)

Installation

1. Clone or set up the project:

   cd /path/to/your/project

2. Create and activate virtual environment:

   python -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate

3. Install dependencies:

   pip install -r requirements.txt

4. Set up environment variables:

   Create a .env file in the project root:

   DATABASE_URL=postgresql+asyncpg://username:password@localhost:5432/your_database_name
   APP_ENV=development
   DEBUG=True

5. Create database:

   Make sure PostgreSQL is running and create a database for the application.

6. Run database migrations (optional, tables are created automatically):

   # Tables will be created automatically on startup

Running the Application

Development

python main.py

The API will be available at:

• API: http://localhost:8000
• API Docs: http://localhost:8000/docs
• ReDoc: http://localhost:8000/redoc

Production

Use a production ASGI server:

uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

API Endpoints

Users

• POST /users/ - Create a new user
• GET /users/ - Get all users
• GET /users/{user_id} - Get user by user_id
• PUT /users/{user_id} - Update user
• DELETE /users/{user_id} - Delete user
• POST /users/authenticate - Authenticate user

Example API Usage

Create a User

curl -X POST "http://localhost:8000/users/" \
     -H "Content-Type: application/json" \
     -d '{"user_id": "user123", "token": "abc123token"}'

Get All Users

curl -X GET "http://localhost:8000/users/"

Authenticate User

curl -X POST "http://localhost:8000/users/authenticate" \
     -H "Content-Type: application/x-www-form-urlencoded" \
     -d "user_id=user123&token=abc123token"

WebSocket API

WebSocket Endpoints

• WS /ws/connect?token=<user_token> - Main WebSocket connection
• WS /ws/echo?token=<user_token> - Echo WebSocket for testing

REST Endpoints for Messages

• GET /messages/user/{user_id} - Get messages for a user
• GET /messages/type/{message_type} - Get messages by type
• GET /messages/ - Get all messages

WebSocket Message Format

Messages should be sent as JSON:

{
  "type": "message_type",
  "payload": {
    "key": "value"
  }
}

Supported Message Types

• ping - Ping/pong heartbeat
• user_data - User data payload
• broadcast - Broadcast message to all users

WebSocket Examples

Connect and Send User Data

// JavaScript WebSocket client example
const ws = new WebSocket('ws://localhost:8000/ws/connect?token=your_user_token');

ws.onopen = function(event) {
    console.log('Connected to WebSocket');

    // Send user data
    ws.send(JSON.stringify({
        type: 'user_data',
        payload: {
            location: 'New York',
            status: 'online',
            customData: 'any data here'
        }
    }));
};

ws.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log('Received:', data);
};

ws.onclose = function(event) {
    console.log('Connection closed');
};

Ping/Pong

ws.send(JSON.stringify({
    type: 'ping',
    payload: { timestamp: Date.now() }
}));

Retrieve Stored Messages

# Get messages for a specific user
curl "http://localhost:8000/messages/user/user123"

# Get messages by type
curl "http://localhost:8000/messages/type/user_data"

# Get all messages
curl "http://localhost:8000/messages/"

Quiz System

Quiz Data Models

Question Model:

• id: Primary key
• text: Question text
• option1, option2, option3, option4: Answer options
• correct_option: Correct answer (1-4)

Quiz Model:

• id: Primary key
• first_user_id: First user ID
• first_user_score: First user score
• second_user_id: Second user ID
• second_user_score: Second user score

Question Management

Create Questions

# Create a question
curl -X POST "http://localhost:8000/questions/" \
     -H "Content-Type: application/json" \
     -d '{
       "text": "What is the capital of France?",
       "option1": "London",
       "option2": "Paris",
       "option3": "Berlin",
       "option4": "Madrid",
       "correct_option": 2
     }'

Get Questions

# Get all questions
curl "http://localhost:8000/questions/"

# Get random questions
curl "http://localhost:8000/questions/random/5"

# Get specific question
curl "http://localhost:8000/questions/1"

Quiz Sessions

Start a Quiz Session

# Start quiz session between two users
curl -X POST "http://localhost:8000/quizzes/start-session?first_user_id=user1&second_user_id=user2"

Response:

{
  "session_id": "uuid-here",
  "first_user_id": "user1",
  "second_user_id": "user2",
  "status": "waiting_for_connections",
  "message": "Quiz session created. Both users should connect via WebSocket."
}

Quiz WebSocket Flow

1. Start Session: Create session via REST API
2. Connect Users: Both users connect via WebSocket
3. Quiz Begins: 10 random questions sent one by one
4. Answer Race: First user to answer gets the point
5. Score Tracking: Scores updated in real-time
6. Completion: Final scores saved and sent to users

WebSocket Quiz Connection

// User 1 connects
const ws1 = new WebSocket('ws://localhost:8000/ws/quiz/play?token=user1_token&session_id=session_uuid');

// User 2 connects
const ws2 = new WebSocket('ws://localhost:8000/ws/quiz/play?token=user2_token&session_id=session_uuid');

ws1.onmessage = (event) => {
    const data = JSON.parse(event.data);

    if (data.type === 'question') {
        // Send answer immediately
        ws1.send(JSON.stringify({
            user_token: 'user1_token',
            question_id: data.question.id,
            answer_number: 2  // Your answer
        }));
    }

    if (data.type === 'quiz_complete') {
        console.log('Quiz finished!', data.results);
    }
};

WebSocket Message Types

• connected: User connected successfully
• quiz_starting: Quiz about to begin
• question: New question with options
• answer_result: Answer processed with scores
• quiz_complete: Final results with winner

Get Quiz Results

# Get all quizzes
curl "http://localhost:8000/quizzes/"

# Get user's quizzes
curl "http://localhost:8000/quizzes/user/user1"

# Get quiz result with winner
curl "http://localhost:8000/quizzes/1/result"

# Get active sessions
curl "http://localhost:8000/quizzes/sessions/active"

Project Structure

├── api/                 # API routes
│   └── users.py        # User management endpoints
├── config/             # Configuration files
│   └── database.py     # Database configuration
├── models/             # SQLAlchemy models
│   ├── base.py         # Base model class
│   └── user.py         # User model
├── schemas/            # Pydantic schemas
│   └── user.py         # User validation schemas
├── services/           # Business logic
│   └── user_service.py # User service layer
├── tests/              # Test files
├── workers/            # Background workers
├── main.py             # FastAPI application
├── requirements.txt    # Python dependencies
└── README.md           # This file

Testing

Run tests with pytest:

pytest

Run with coverage:

pytest --cov=.

Development Guidelines

• Type Hints: All functions must have type hints
• Async/Await: Prefer async operations where possible
• PEP 8: Follow Python style guidelines
• F-strings: Use f-strings for string formatting
• Docstrings: Add docstrings to all public functions and classes
• Error Handling: Proper exception handling and meaningful error messages
• Validation: Validate all input data with Pydantic

License

This project is open source. Feel free to use and modify as needed.