nice
This commit is contained in:
@@ -0,0 +1,6 @@
|
||||
# API Requirements for Mice Game Score API
|
||||
fastapi>=0.104.1
|
||||
uvicorn[standard]>=0.24.0
|
||||
pydantic>=2.5.0
|
||||
sqlite3 # Built into Python
|
||||
python-multipart>=0.0.6
|
||||
Binary file not shown.
@@ -0,0 +1,547 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Mice Game Score API
|
||||
FastAPI application with SQLite database for user and score management
|
||||
"""
|
||||
|
||||
from fastapi import FastAPI, HTTPException, Depends
|
||||
from fastapi.responses import JSONResponse
|
||||
import sqlite3
|
||||
import os
|
||||
from datetime import datetime
|
||||
from typing import List, Optional
|
||||
from pydantic import BaseModel
|
||||
import hashlib
|
||||
import uuid
|
||||
|
||||
|
||||
# Pydantic models for request/response
|
||||
class User(BaseModel):
|
||||
user_id: str
|
||||
device_id: str
|
||||
created_at: Optional[str] = None
|
||||
last_active: Optional[str] = None
|
||||
|
||||
class Score(BaseModel):
|
||||
user_id: str
|
||||
device_id: str
|
||||
score: int
|
||||
game_completed: bool = True
|
||||
timestamp: Optional[str] = None
|
||||
|
||||
class ScoreResponse(BaseModel):
|
||||
id: int
|
||||
user_id: str
|
||||
device_id: str
|
||||
score: int
|
||||
game_completed: bool
|
||||
timestamp: str
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
user_id: str
|
||||
device_id: str
|
||||
created_at: str
|
||||
last_active: str
|
||||
total_scores: int
|
||||
best_score: int
|
||||
|
||||
|
||||
# Database setup
|
||||
DATABASE_FILE = "mice_game.db"
|
||||
|
||||
def init_database():
|
||||
"""Initialize the SQLite database with required tables"""
|
||||
conn = sqlite3.connect(DATABASE_FILE)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Create users table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS users (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
created_at TEXT NOT NULL,
|
||||
last_active TEXT NOT NULL,
|
||||
UNIQUE(user_id, device_id)
|
||||
)
|
||||
''')
|
||||
|
||||
# Create scores table
|
||||
cursor.execute('''
|
||||
CREATE TABLE IF NOT EXISTS scores (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
user_id TEXT NOT NULL,
|
||||
device_id TEXT NOT NULL,
|
||||
score INTEGER NOT NULL,
|
||||
game_completed BOOLEAN NOT NULL DEFAULT 1,
|
||||
timestamp TEXT NOT NULL,
|
||||
FOREIGN KEY (user_id, device_id) REFERENCES users (user_id, device_id)
|
||||
)
|
||||
''')
|
||||
|
||||
# Create indexes for better performance
|
||||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_users_device ON users(device_id)')
|
||||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_user_device ON scores(user_id, device_id)')
|
||||
cursor.execute('CREATE INDEX IF NOT EXISTS idx_scores_timestamp ON scores(timestamp)')
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
def get_db_connection():
|
||||
"""Get database connection"""
|
||||
if not os.path.exists(DATABASE_FILE):
|
||||
init_database()
|
||||
|
||||
conn = sqlite3.connect(DATABASE_FILE)
|
||||
conn.row_factory = sqlite3.Row # Enable column access by name
|
||||
return conn
|
||||
|
||||
def close_db_connection(conn):
|
||||
"""Close database connection"""
|
||||
conn.close()
|
||||
|
||||
|
||||
# FastAPI app
|
||||
app = FastAPI(
|
||||
title="Mice Game Score API",
|
||||
description="API for managing users and scores in the Mice game",
|
||||
version="1.0.0"
|
||||
)
|
||||
|
||||
# Initialize database on startup
|
||||
@app.on_event("startup")
|
||||
def startup_event():
|
||||
init_database()
|
||||
|
||||
|
||||
# Utility functions
|
||||
def validate_device_id(device_id: str) -> bool:
|
||||
"""Validate device ID format"""
|
||||
return device_id.startswith("DEV-") and len(device_id) == 12
|
||||
|
||||
def validate_user_id(user_id: str) -> bool:
|
||||
"""Validate user ID format"""
|
||||
return len(user_id.strip()) > 0 and len(user_id) <= 50
|
||||
|
||||
def user_exists(conn, user_id: str, device_id: str) -> bool:
|
||||
"""Check if user exists in database"""
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"SELECT 1 FROM users WHERE user_id = ? AND device_id = ?",
|
||||
(user_id, device_id)
|
||||
)
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
def user_id_unique_globally(conn, user_id: str) -> bool:
|
||||
"""Check if user_id is globally unique (across all devices)"""
|
||||
cursor = conn.cursor()
|
||||
cursor.execute("SELECT 1 FROM users WHERE user_id = ?", (user_id,))
|
||||
return cursor.fetchone() is None
|
||||
|
||||
|
||||
# API Endpoints
|
||||
|
||||
@app.get("/")
|
||||
def read_root():
|
||||
"""API root endpoint"""
|
||||
return {
|
||||
"message": "Mice Game Score API",
|
||||
"version": "1.0.0",
|
||||
"endpoints": [
|
||||
"GET /users/{device_id} - Get all users for device",
|
||||
"POST /signup/{device_id}/{user_id} - Register new user",
|
||||
"POST /score/{device_id}/{user_id} - Submit score"
|
||||
]
|
||||
}
|
||||
|
||||
@app.post("/signup/{device_id}/{user_id}")
|
||||
def signup_user(device_id: str, user_id: str):
|
||||
"""
|
||||
Register a new user for a device
|
||||
- device_id: Device identifier (format: DEV-XXXXXXXX)
|
||||
- user_id: Unique user identifier
|
||||
"""
|
||||
# Validate inputs
|
||||
if not validate_device_id(device_id):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
||||
)
|
||||
|
||||
if not validate_user_id(user_id):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Invalid user_id. Must be 1-50 characters long"
|
||||
)
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
# Check if user_id is globally unique
|
||||
if not user_id_unique_globally(conn, user_id):
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"User ID '{user_id}' already exists. Choose a different user ID."
|
||||
)
|
||||
|
||||
# Check if user already exists for this device
|
||||
if user_exists(conn, user_id, device_id):
|
||||
raise HTTPException(
|
||||
status_code=409,
|
||||
detail=f"User '{user_id}' already registered for device '{device_id}'"
|
||||
)
|
||||
|
||||
# Register the user
|
||||
cursor = conn.cursor()
|
||||
now = datetime.now().isoformat()
|
||||
|
||||
cursor.execute(
|
||||
"INSERT INTO users (user_id, device_id, created_at, last_active) VALUES (?, ?, ?, ?)",
|
||||
(user_id, device_id, now, now)
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"User '{user_id}' successfully registered for device '{device_id}'",
|
||||
"user": {
|
||||
"user_id": user_id,
|
||||
"device_id": device_id,
|
||||
"created_at": now
|
||||
}
|
||||
}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
@app.post("/score/{device_id}/{user_id}")
|
||||
def submit_score(device_id: str, user_id: str, score_data: Score):
|
||||
"""
|
||||
Submit a score for a registered user
|
||||
- device_id: Device identifier
|
||||
- user_id: User identifier
|
||||
- score_data: Score information (score, game_completed)
|
||||
"""
|
||||
# Validate inputs
|
||||
if not validate_device_id(device_id):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
||||
)
|
||||
|
||||
if not validate_user_id(user_id):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Invalid user_id"
|
||||
)
|
||||
|
||||
if score_data.score < 0:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Score must be non-negative"
|
||||
)
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
# Check if user exists and is registered for this device
|
||||
if not user_exists(conn, user_id, device_id):
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"User '{user_id}' not registered for device '{device_id}'. Please signup first."
|
||||
)
|
||||
|
||||
# Add the score
|
||||
cursor = conn.cursor()
|
||||
timestamp = datetime.now().isoformat()
|
||||
|
||||
cursor.execute(
|
||||
"INSERT INTO scores (user_id, device_id, score, game_completed, timestamp) VALUES (?, ?, ?, ?, ?)",
|
||||
(user_id, device_id, score_data.score, score_data.game_completed, timestamp)
|
||||
)
|
||||
|
||||
# Update user's last_active timestamp
|
||||
cursor.execute(
|
||||
"UPDATE users SET last_active = ? WHERE user_id = ? AND device_id = ?",
|
||||
(timestamp, user_id, device_id)
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
|
||||
# Get user's statistics
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
COUNT(*) as total_games,
|
||||
MAX(score) as best_score,
|
||||
AVG(score) as avg_score,
|
||||
SUM(score) as total_score
|
||||
FROM scores
|
||||
WHERE user_id = ? AND device_id = ?
|
||||
""",
|
||||
(user_id, device_id)
|
||||
)
|
||||
|
||||
stats = cursor.fetchone()
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Score {score_data.score} submitted successfully",
|
||||
"score_id": cursor.lastrowid,
|
||||
"user_stats": {
|
||||
"user_id": user_id,
|
||||
"device_id": device_id,
|
||||
"total_games": stats["total_games"],
|
||||
"best_score": stats["best_score"],
|
||||
"average_score": round(stats["avg_score"], 2) if stats["avg_score"] else 0,
|
||||
"total_score": stats["total_score"]
|
||||
}
|
||||
}
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
@app.get("/users/{device_id}")
|
||||
def get_device_users(device_id: str) -> List[UserResponse]:
|
||||
"""
|
||||
Get all registered users for a specific device
|
||||
- device_id: Device identifier
|
||||
"""
|
||||
# Validate device_id
|
||||
if not validate_device_id(device_id):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Invalid device_id format. Must be 'DEV-XXXXXXXX'"
|
||||
)
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Get users with their statistics
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
u.user_id,
|
||||
u.device_id,
|
||||
u.created_at,
|
||||
u.last_active,
|
||||
COUNT(s.id) as total_scores,
|
||||
COALESCE(MAX(s.score), 0) as best_score
|
||||
FROM users u
|
||||
LEFT JOIN scores s ON u.user_id = s.user_id AND u.device_id = s.device_id
|
||||
WHERE u.device_id = ?
|
||||
GROUP BY u.user_id, u.device_id, u.created_at, u.last_active
|
||||
ORDER BY u.last_active DESC
|
||||
""",
|
||||
(device_id,)
|
||||
)
|
||||
|
||||
users = cursor.fetchall()
|
||||
|
||||
if not users:
|
||||
return []
|
||||
|
||||
return [
|
||||
UserResponse(
|
||||
user_id=user["user_id"],
|
||||
device_id=user["device_id"],
|
||||
created_at=user["created_at"],
|
||||
last_active=user["last_active"],
|
||||
total_scores=user["total_scores"],
|
||||
best_score=user["best_score"]
|
||||
)
|
||||
for user in users
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
# Additional useful endpoints
|
||||
|
||||
@app.get("/scores/{device_id}/{user_id}")
|
||||
def get_user_scores(device_id: str, user_id: str, limit: int = 10) -> List[ScoreResponse]:
|
||||
"""
|
||||
Get recent scores for a user
|
||||
- device_id: Device identifier
|
||||
- user_id: User identifier
|
||||
- limit: Maximum number of scores to return (default: 10)
|
||||
"""
|
||||
if not validate_device_id(device_id) or not validate_user_id(user_id):
|
||||
raise HTTPException(status_code=400, detail="Invalid device_id or user_id")
|
||||
|
||||
if limit < 1 or limit > 100:
|
||||
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
# Check if user exists
|
||||
if not user_exists(conn, user_id, device_id):
|
||||
raise HTTPException(
|
||||
status_code=404,
|
||||
detail=f"User '{user_id}' not found for device '{device_id}'"
|
||||
)
|
||||
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT id, user_id, device_id, score, game_completed, timestamp
|
||||
FROM scores
|
||||
WHERE user_id = ? AND device_id = ?
|
||||
ORDER BY timestamp DESC
|
||||
LIMIT ?
|
||||
""",
|
||||
(user_id, device_id, limit)
|
||||
)
|
||||
|
||||
scores = cursor.fetchall()
|
||||
|
||||
return [
|
||||
ScoreResponse(
|
||||
id=score["id"],
|
||||
user_id=score["user_id"],
|
||||
device_id=score["device_id"],
|
||||
score=score["score"],
|
||||
game_completed=bool(score["game_completed"]),
|
||||
timestamp=score["timestamp"]
|
||||
)
|
||||
for score in scores
|
||||
]
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
@app.get("/leaderboard/{device_id}")
|
||||
def get_device_leaderboard(device_id: str, limit: int = 10):
|
||||
"""
|
||||
Get leaderboard for a device (top scores)
|
||||
- device_id: Device identifier
|
||||
- limit: Maximum number of entries to return (default: 10)
|
||||
"""
|
||||
if not validate_device_id(device_id):
|
||||
raise HTTPException(status_code=400, detail="Invalid device_id")
|
||||
|
||||
if limit < 1 or limit > 100:
|
||||
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
user_id,
|
||||
MAX(score) as best_score,
|
||||
COUNT(*) as total_games,
|
||||
MAX(timestamp) as last_play
|
||||
FROM scores
|
||||
WHERE device_id = ?
|
||||
GROUP BY user_id
|
||||
ORDER BY best_score DESC, last_play DESC
|
||||
LIMIT ?
|
||||
""",
|
||||
(device_id, limit)
|
||||
)
|
||||
|
||||
leaderboard = cursor.fetchall()
|
||||
|
||||
return [
|
||||
{
|
||||
"rank": idx + 1,
|
||||
"user_id": entry["user_id"],
|
||||
"best_score": entry["best_score"],
|
||||
"total_games": entry["total_games"],
|
||||
"last_play": entry["last_play"]
|
||||
}
|
||||
for idx, entry in enumerate(leaderboard)
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
|
||||
@app.get("/leaderboard/global/top")
|
||||
def get_global_leaderboard(limit: int = 10):
|
||||
"""
|
||||
Get global leaderboard across all devices (absolute top scores)
|
||||
- limit: Maximum number of entries to return (default: 10)
|
||||
"""
|
||||
if limit < 1 or limit > 100:
|
||||
raise HTTPException(status_code=400, detail="Limit must be between 1 and 100")
|
||||
|
||||
conn = get_db_connection()
|
||||
|
||||
try:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT
|
||||
user_id,
|
||||
device_id,
|
||||
MAX(score) as best_score,
|
||||
COUNT(*) as total_games,
|
||||
MAX(timestamp) as last_play
|
||||
FROM scores
|
||||
GROUP BY user_id, device_id
|
||||
ORDER BY best_score DESC, last_play DESC
|
||||
LIMIT ?
|
||||
""",
|
||||
(limit,)
|
||||
)
|
||||
|
||||
leaderboard = cursor.fetchall()
|
||||
|
||||
return [
|
||||
{
|
||||
"rank": idx + 1,
|
||||
"user_id": entry["user_id"],
|
||||
"device_id": entry["device_id"],
|
||||
"best_score": entry["best_score"],
|
||||
"total_games": entry["total_games"],
|
||||
"last_play": entry["last_play"]
|
||||
}
|
||||
for idx, entry in enumerate(leaderboard)
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")
|
||||
finally:
|
||||
close_db_connection(conn)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
|
||||
print("Starting Mice Game Score API...")
|
||||
print("Available endpoints:")
|
||||
print(" POST /signup/{device_id}/{user_id} - Register new user")
|
||||
print(" POST /score/{device_id}/{user_id} - Submit score")
|
||||
print(" GET /users/{device_id} - Get all users for device")
|
||||
print(" GET /scores/{device_id}/{user_id} - Get user scores")
|
||||
print(" GET /leaderboard/{device_id} - Get device leaderboard")
|
||||
print()
|
||||
|
||||
# Initialize database
|
||||
init_database()
|
||||
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
Reference in New Issue
Block a user