b30e54a8f7
Full-stack app for participatory water pricing using Bezier curves. - Backend: FastAPI + SQLAlchemy + SQLite with JWT auth - Frontend: Nuxt 4 + TypeScript with interactive SVG editor - Math engine: cubic Bezier tarification with Cardano solver - Admin: commune management, household import, vote monitoring, CMS - Citizen: interactive curve editor, vote submission - Docker-compose deployment ready Includes fixes for: - Impact table snake_case/camelCase property mismatch - CMS content backend API + frontend editor (was stub) - Admin route protection middleware - Public content display on commune page - Vote confirmation page link fix Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.database import get_db
|
|
from app.models import AdminUser, Household, Commune
|
|
from app.schemas import AdminLogin, CitizenVerify, Token, AdminUserCreate, AdminUserOut
|
|
from app.services.auth_service import (
|
|
verify_password, create_admin_token, create_citizen_token,
|
|
hash_password, require_super_admin,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/admin/login", response_model=Token)
|
|
async def admin_login(data: AdminLogin, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(
|
|
select(AdminUser)
|
|
.options(selectinload(AdminUser.communes))
|
|
.where(AdminUser.email == data.email)
|
|
)
|
|
admin = result.scalar_one_or_none()
|
|
if not admin or not verify_password(data.password, admin.hashed_password):
|
|
raise HTTPException(status_code=401, detail="Identifiants invalides")
|
|
|
|
# For commune_admin, include their first commune slug
|
|
commune_slug = None
|
|
if admin.communes:
|
|
commune_slug = admin.communes[0].slug
|
|
|
|
return Token(
|
|
access_token=create_admin_token(admin),
|
|
role=admin.role,
|
|
commune_slug=commune_slug,
|
|
)
|
|
|
|
|
|
@router.post("/citizen/verify", response_model=Token)
|
|
async def citizen_verify(data: CitizenVerify, db: AsyncSession = Depends(get_db)):
|
|
result = await db.execute(
|
|
select(Household)
|
|
.join(Commune)
|
|
.where(Commune.slug == data.commune_slug, Household.auth_code == data.auth_code)
|
|
)
|
|
household = result.scalar_one_or_none()
|
|
if not household:
|
|
raise HTTPException(status_code=401, detail="Code invalide ou commune introuvable")
|
|
return Token(
|
|
access_token=create_citizen_token(household, data.commune_slug),
|
|
role="citizen",
|
|
commune_slug=data.commune_slug,
|
|
)
|
|
|
|
|
|
@router.post("/admin/create", response_model=AdminUserOut)
|
|
async def create_admin(
|
|
data: AdminUserCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current: AdminUser = Depends(require_super_admin),
|
|
):
|
|
existing = await db.execute(select(AdminUser).where(AdminUser.email == data.email))
|
|
if existing.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="Email déjà utilisé")
|
|
|
|
admin = AdminUser(
|
|
email=data.email,
|
|
hashed_password=hash_password(data.password),
|
|
full_name=data.full_name,
|
|
role=data.role,
|
|
)
|
|
|
|
if data.commune_slugs:
|
|
for slug in data.commune_slugs:
|
|
result = await db.execute(select(Commune).where(Commune.slug == slug))
|
|
commune = result.scalar_one_or_none()
|
|
if commune:
|
|
admin.communes.append(commune)
|
|
|
|
db.add(admin)
|
|
await db.commit()
|
|
await db.refresh(admin)
|
|
return admin
|