新增小程序后端代码,包括数据库、路由、服务层等。

小程序前后端打通,包括登录、注册、绑定设备、查询绑定信息,修改小朋友名称等功能。
This commit is contained in:
HycJack
2026-04-13 01:54:57 +08:00
parent 33715373b0
commit a22fcafea9
46 changed files with 3713 additions and 268 deletions

View File

@@ -0,0 +1,164 @@
import logging
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel
try:
from app.security import get_current_user_id
from app.service.binding import BindingService
from app.service import get_db_session
except ModuleNotFoundError:
from security import get_current_user_id
from service.binding import BindingService
from service import get_db_session
router = APIRouter(prefix="/bindings", tags=["bindings"])
logger = logging.getLogger("app.bindings")
class BindStartRequest(BaseModel):
device_id: str
child_id: int
class BindStartResponse(BaseModel):
bind_token: str
expires_at: str
class BindConfirmRequest(BaseModel):
bind_token: str
challenge_code: str
class BindConfirmResponse(BaseModel):
device_id: str
child_id: int
class BindingGetResponse(BaseModel):
device_id: str
child_id: int
status: int
bound_at: str
class BindHistoryItem(BaseModel):
device_id: str
child_id: int
bound_at: str
unbound_at: str | None
class BindHistoryResponse(BaseModel):
items: list[BindHistoryItem]
total: int
next_cursor: str | None
@router.post("/start", response_model=BindStartResponse)
def start_bind(
payload: BindStartRequest,
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> BindStartResponse:
service = BindingService(db)
bind_token, expires_at = service.start_bind(current_user_id, payload.device_id, payload.child_id)
return BindStartResponse(bind_token=bind_token, expires_at=expires_at.isoformat())
@router.post("/confirm", response_model=BindConfirmResponse)
def confirm_bind(
payload: BindConfirmRequest,
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> BindConfirmResponse:
service = BindingService(db)
try:
result = service.confirm_bind(payload.bind_token, current_user_id)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return BindConfirmResponse(**result)
class DirectBindRequest(BaseModel):
device_id: str
child_id: int
class DirectBindResponse(BaseModel):
device_id: str
child_id: int
@router.post("/direct", response_model=DirectBindResponse)
def direct_bind(
payload: DirectBindRequest,
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> DirectBindResponse:
service = BindingService(db)
result = service.direct_bind(payload.device_id, payload.child_id, current_user_id)
return DirectBindResponse(**result)
@router.get("/current", response_model=BindingGetResponse)
def get_current_binding(
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
):
service = BindingService(db)
binding = service.get_current_binding(current_user_id)
if not binding:
raise HTTPException(status_code=404, detail="no binding found")
return binding
@router.get("/{device_id}", response_model=BindingGetResponse)
def get_binding(
device_id: str,
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> BindingGetResponse:
service = BindingService(db)
binding = service.get_binding(device_id, current_user_id)
if not binding:
raise HTTPException(status_code=404, detail="binding not found")
return BindingGetResponse(**binding)
@router.delete("/{device_id}", status_code=status.HTTP_204_NO_CONTENT)
def unbind_device(
device_id: str,
request: Request,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> None:
service = BindingService(db)
if not service.unbind(device_id, current_user_id):
raise HTTPException(status_code=404, detail="binding not found")
@router.get("/history/{device_id}", response_model=BindHistoryResponse)
def get_bind_history(
device_id: str,
request: Request,
cursor: str | None = None,
limit: int = 20,
current_user_id: int = Depends(get_current_user_id),
db=Depends(get_db_session),
) -> BindHistoryResponse:
from datetime import datetime
cursor_dt = datetime.fromisoformat(cursor) if cursor else None
service = BindingService(db)
rows, has_more = service.list_history(device_id, limit, cursor_dt)
next_cursor = rows[-1]["bound_at"].isoformat() if has_more and rows else None
items = [BindHistoryItem(**row) for row in rows]
return BindHistoryResponse(items=items, total=len(items), next_cursor=next_cursor)