98 lines
3.3 KiB
Python
98 lines
3.3 KiB
Python
import json
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
|
from sqlalchemy.orm import Session as DbSession
|
|
|
|
from app.auth import verify_api_key
|
|
from app.database import get_db
|
|
from app.models import ExternalData, Session, User
|
|
from app.schemas import ExternalDataCreate, ExternalDataPublic, SessionPublic, UserPublic
|
|
|
|
router = APIRouter(prefix="/api/v1", tags=["external-api"])
|
|
|
|
|
|
@router.get("/users/{user_id}", response_model=UserPublic)
|
|
def get_user(
|
|
user_id: int,
|
|
_: None = Depends(verify_api_key),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
return user
|
|
|
|
|
|
@router.get("/users/by-email/{email}", response_model=UserPublic)
|
|
def get_user_by_email(
|
|
email: str,
|
|
_: None = Depends(verify_api_key),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
user = db.query(User).filter(User.email == email.strip().lower()).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
return user
|
|
|
|
|
|
@router.get("/users/{user_id}/sessions", response_model=list[SessionPublic])
|
|
def get_user_sessions(
|
|
user_id: int,
|
|
active_only: bool = True,
|
|
_: None = Depends(verify_api_key),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
query = db.query(Session).filter(Session.user_id == user_id)
|
|
if active_only:
|
|
query = query.filter(Session.is_active.is_(True))
|
|
return query.order_by(Session.created_at.desc()).all()
|
|
|
|
|
|
@router.post("/users/{user_id}/data", response_model=ExternalDataPublic, status_code=status.HTTP_201_CREATED)
|
|
def write_external_data(
|
|
user_id: int,
|
|
body: ExternalDataCreate,
|
|
_: None = Depends(verify_api_key),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
|
|
payload_str = body.payload if isinstance(body.payload, str) else json.dumps(body.payload, ensure_ascii=False)
|
|
|
|
record = ExternalData(user_id=user_id, source=body.source, payload=payload_str)
|
|
db.add(record)
|
|
|
|
if body.points_delta:
|
|
user.points = max(0, user.points + body.points_delta)
|
|
if body.rating_delta:
|
|
user.rating = max(0.0, user.rating + body.rating_delta)
|
|
|
|
db.commit()
|
|
db.refresh(record)
|
|
return record
|
|
|
|
|
|
@router.get("/users/{user_id}/data", response_model=list[ExternalDataPublic])
|
|
def get_external_data(
|
|
user_id: int,
|
|
limit: int = 50,
|
|
_: None = Depends(verify_api_key),
|
|
db: DbSession = Depends(get_db),
|
|
):
|
|
user = db.query(User).filter(User.id == user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
limit = min(max(limit, 1), 100)
|
|
return (
|
|
db.query(ExternalData)
|
|
.filter(ExternalData.user_id == user_id)
|
|
.order_by(ExternalData.created_at.desc())
|
|
.limit(limit)
|
|
.all()
|
|
)
|