File size: 5,471 Bytes
4ec5660 5ae58b5 4ec5660 5ae58b5 4ec5660 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 |
import os
import logging
import aiohttp
from fastapi import APIRouter, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
router = APIRouter()
# 🔧 Supabase Config
SUPABASE_URL = "https://ussxqnifefkgkaumjann.supabase.co"
SUPABASE_KEY = os.getenv("SUPA_KEY")
SUPABASE_ROLE_KEY = os.getenv("SUPA_SERVICE_KEY")
if not SUPABASE_KEY or not SUPABASE_ROLE_KEY:
raise ValueError("❌ SUPA_KEY or SUPA_SERVICE_KEY not set in environment!")
SUPABASE_HEADERS = {
"apikey": SUPABASE_KEY,
"Authorization": f"Bearer {SUPABASE_KEY}",
"Content-Type": "application/json"
}
SUPABASE_ROLE_HEADERS = {
"apikey": SUPABASE_ROLE_KEY,
"Authorization": f"Bearer {SUPABASE_ROLE_KEY}",
"Content-Type": "application/json"
}
class RemoveCollaborationRequest(BaseModel):
user_id: str
# 🔍 Model
class CreateCollaborationRequest(BaseModel):
email: str
password: str
# ✅ Verifica se o token pertence a um admin
async def verify_admin_token(user_token: str) -> str:
headers = {
"Authorization": f"Bearer {user_token}",
"apikey": SUPABASE_KEY,
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session:
async with session.get(f"{SUPABASE_URL}/auth/v1/user", headers=headers) as response:
if response.status != 200:
raise HTTPException(status_code=401, detail="Invalid or expired token")
user_data = await response.json()
user_id = user_data.get("id")
if not user_id:
raise HTTPException(status_code=400, detail="User ID not found")
# Verifica se é admin na tabela User
async with aiohttp.ClientSession() as session:
async with session.get(
f"{SUPABASE_URL}/rest/v1/User?id=eq.{user_id}",
headers=SUPABASE_HEADERS
) as resp:
if resp.status != 200:
raise HTTPException(status_code=403, detail="Could not verify admin status")
user_info = await resp.json()
if not user_info or not user_info[0].get("is_admin", False):
raise HTTPException(status_code=403, detail="Admin privileges required")
return user_id
@router.post("/admin/remove-collaboration")
async def remove_collaboration(
body: RemoveCollaborationRequest,
user_token: str = Header(None, alias="User-key")
):
try:
await verify_admin_token(user_token)
update_payload = {
"is_admin": False
}
async with aiohttp.ClientSession() as session:
async with session.patch(
f"{SUPABASE_URL}/rest/v1/User?id=eq.{body.user_id}",
headers=SUPABASE_ROLE_HEADERS,
json=update_payload
) as response:
if response.status != 204:
error_detail = await response.text()
raise HTTPException(status_code=500, detail=f"Failed to update user admin status: {error_detail}")
return {"success": True, "message": "Colaborador removido com sucesso"}
except HTTPException as http_ex:
return {"success": False, "detail": http_ex.detail}
except Exception as e:
logging.error(f"❌ Unexpected error: {str(e)}")
return {"success": False, "detail": str(e)}
# ➕ Criar colaborador
@router.post("/admin/create-collaboration")
async def create_collaboration(
body: CreateCollaborationRequest,
user_token: str = Header(None, alias="User-key")
):
try:
await verify_admin_token(user_token)
# 1. Cria o novo usuário com Admin API
create_user_payload = {
"email": body.email,
"password": body.password,
"email_confirm": True
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{SUPABASE_URL}/auth/v1/admin/users",
headers=SUPABASE_ROLE_HEADERS,
json=create_user_payload
) as user_response:
if user_response.status != 200:
error_detail = await user_response.json()
raise HTTPException(status_code=400, detail=error_detail.get("msg", "Failed to create user"))
user_data = await user_response.json()
user_id = user_data.get("id")
if not user_id:
raise HTTPException(status_code=500, detail="User ID not returned after creation")
# 2. Cria o registro na tabela "User" com is_admin = true
user_insert_payload = {
"id": user_id,
"email": body.email,
"is_admin": True
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{SUPABASE_URL}/rest/v1/User",
headers=SUPABASE_ROLE_HEADERS,
json=user_insert_payload
) as insert_response:
if insert_response.status != 201:
error_detail = await insert_response.text()
raise HTTPException(status_code=500, detail="Failed to insert user into User table")
return {"success": True}
except HTTPException as http_ex:
return {"success": False, "detail": http_ex.detail}
except Exception as e:
logging.error(f"❌ Unexpected error: {str(e)}")
return {"success": False, "detail": str(e)} |