130 lines
3.7 KiB
Python
130 lines
3.7 KiB
Python
import sqlite3
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, HTTPException, status
|
|
|
|
from app.database import Database
|
|
from app.schemas import User, UserCreate, UserUpdate
|
|
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
def row_to_user(row: sqlite3.Row) -> User:
|
|
return User(
|
|
id=row["id"],
|
|
registered_at=row["registered_at"],
|
|
name=row["name"],
|
|
phone=row["phone"],
|
|
nickname=row["nickname"],
|
|
avatar=row["avatar"],
|
|
is_admin=bool(row["is_admin"]),
|
|
total_spent=float(row["total_spent"]),
|
|
)
|
|
|
|
|
|
def raise_duplicate_user_error() -> None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_409_CONFLICT,
|
|
detail="Phone already exists",
|
|
)
|
|
|
|
|
|
@router.get("", response_model=List[User])
|
|
def list_users(db: Database, phone: str | None = None) -> List[User]:
|
|
params = []
|
|
where_clause = ""
|
|
if phone:
|
|
where_clause = "WHERE phone LIKE ?"
|
|
params.append(f"%{phone}%")
|
|
|
|
rows = db.execute(
|
|
f"""
|
|
SELECT id, registered_at, name, phone, nickname, avatar, is_admin, total_spent
|
|
FROM users
|
|
{where_clause}
|
|
ORDER BY id
|
|
""",
|
|
params,
|
|
).fetchall()
|
|
return [row_to_user(row) for row in rows]
|
|
|
|
|
|
@router.post("", response_model=User, status_code=status.HTTP_201_CREATED)
|
|
def create_user(payload: UserCreate, db: Database) -> User:
|
|
try:
|
|
cursor = db.execute(
|
|
"""
|
|
INSERT INTO users
|
|
(registered_at, name, phone, nickname, avatar, is_admin, total_spent)
|
|
VALUES (CURRENT_TIMESTAMP, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(
|
|
payload.name,
|
|
payload.phone,
|
|
payload.nickname,
|
|
payload.avatar,
|
|
int(payload.is_admin),
|
|
payload.total_spent,
|
|
),
|
|
)
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
raise_duplicate_user_error()
|
|
|
|
return get_user(cursor.lastrowid, db)
|
|
|
|
|
|
@router.get("/{user_id}", response_model=User)
|
|
def get_user(user_id: int, db: Database) -> User:
|
|
row = db.execute(
|
|
"""
|
|
SELECT id, registered_at, name, phone, nickname, avatar, is_admin, total_spent
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|
|
return row_to_user(row)
|
|
|
|
|
|
@router.patch("/{user_id}", response_model=User)
|
|
def update_user(user_id: int, payload: UserUpdate, db: Database) -> User:
|
|
user = get_user(user_id, db)
|
|
updates = payload.model_dump(exclude_unset=True)
|
|
|
|
if not updates:
|
|
return user
|
|
|
|
name = updates.get("name", user.name)
|
|
phone = updates.get("phone", user.phone)
|
|
nickname = updates.get("nickname", user.nickname)
|
|
avatar = updates.get("avatar", user.avatar)
|
|
is_admin = updates.get("is_admin", user.is_admin)
|
|
total_spent = updates.get("total_spent", user.total_spent)
|
|
|
|
try:
|
|
db.execute(
|
|
"""
|
|
UPDATE users
|
|
SET name = ?, phone = ?, nickname = ?, avatar = ?, is_admin = ?, total_spent = ?
|
|
WHERE id = ?
|
|
""",
|
|
(name, phone, nickname, avatar, int(is_admin), total_spent, user_id),
|
|
)
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
raise_duplicate_user_error()
|
|
|
|
return get_user(user_id, db)
|
|
|
|
|
|
@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_user(user_id: int, db: Database) -> None:
|
|
cursor = db.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
db.commit()
|
|
if cursor.rowcount == 0:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
|