821675ab4a
Specs stay at repo root (cross-VM). Move deploy and code into logical projects with README per domain, updated manifest.yaml, and symlinks at legacy paths for VM122 backward compatibility.
115 lines
3.4 KiB
Python
115 lines
3.4 KiB
Python
"""Billing API routes — Spec 023."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
|
|
from app import auth, billing_store
|
|
from app.permissions import can_manage_billing, can_read_billing, should_mask_sensitive
|
|
|
|
router = APIRouter(prefix="/api/v1/billing", tags=["billing"])
|
|
|
|
|
|
class PatchBillingBody(BaseModel):
|
|
billing_state: str | None = None
|
|
recurrence_active: bool | None = None
|
|
external_customer_id: str | None = None
|
|
plan_code: str | None = None
|
|
activated_by: str | None = None
|
|
|
|
|
|
def _reader(user: auth.DeskUser = Depends(auth.get_current_user)) -> auth.DeskUser:
|
|
if not can_read_billing(user.role):
|
|
raise HTTPException(403, "permissão insuficiente")
|
|
return user
|
|
|
|
|
|
def _manager(user: auth.DeskUser = Depends(auth.get_current_user)) -> auth.DeskUser:
|
|
if not can_manage_billing(user.role):
|
|
raise HTTPException(403, "permissão insuficiente")
|
|
return user
|
|
|
|
|
|
@router.get("/summary")
|
|
def billing_summary(user: auth.DeskUser = Depends(_reader)):
|
|
conn = auth.db()
|
|
try:
|
|
data = billing_store.summary(conn)
|
|
if should_mask_sensitive(user.role):
|
|
data["recent_validations"] = [
|
|
billing_store._row_dict(r, mask=True)
|
|
for r in conn.execute(
|
|
"SELECT * FROM billing_accounts ORDER BY updated_at DESC LIMIT 5"
|
|
).fetchall()
|
|
]
|
|
return data
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/accounts")
|
|
def list_billing_accounts(
|
|
billing_state: str = "",
|
|
domain: str = "",
|
|
limit: int = Query(100, ge=1, le=500),
|
|
user: auth.DeskUser = Depends(_reader),
|
|
):
|
|
conn = auth.db()
|
|
try:
|
|
mask = should_mask_sensitive(user.role)
|
|
return billing_store.list_accounts(
|
|
conn,
|
|
billing_state=billing_state.strip() or None,
|
|
domain=domain.strip() or None,
|
|
limit=limit,
|
|
mask=mask,
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.get("/accounts/by-domain/{domain}")
|
|
def billing_by_domain(domain: str, user: auth.DeskUser = Depends(_reader)):
|
|
conn = auth.db()
|
|
try:
|
|
acc = billing_store.get_by_domain(conn, domain, mask=should_mask_sensitive(user.role))
|
|
finally:
|
|
conn.close()
|
|
if not acc:
|
|
raise HTTPException(404, "conta não encontrada")
|
|
return acc
|
|
|
|
|
|
@router.get("/accounts/{account_id}")
|
|
def get_billing_account(account_id: int, user: auth.DeskUser = Depends(_reader)):
|
|
conn = auth.db()
|
|
try:
|
|
acc = billing_store.get_account(conn, account_id, mask=should_mask_sensitive(user.role))
|
|
finally:
|
|
conn.close()
|
|
if not acc:
|
|
raise HTTPException(404, "conta não encontrada")
|
|
return acc
|
|
|
|
|
|
@router.patch("/accounts/{account_id}")
|
|
def patch_billing_account(
|
|
account_id: int,
|
|
body: PatchBillingBody,
|
|
user: auth.DeskUser = Depends(_manager),
|
|
):
|
|
conn = auth.db()
|
|
try:
|
|
fields = body.model_dump(exclude_none=True)
|
|
if body.recurrence_active and not fields.get("activated_by"):
|
|
from datetime import datetime, timezone
|
|
|
|
fields["activated_by"] = user.username
|
|
fields["activated_at"] = datetime.now(timezone.utc).isoformat()
|
|
acc = billing_store.patch_account(conn, account_id, **fields)
|
|
finally:
|
|
conn.close()
|
|
if not acc:
|
|
raise HTTPException(404, "conta não encontrada")
|
|
return acc
|