refactor(api): split views_api.py into a package of domain modules
Pure move — no logic changes. The 4,100-line single file becomes views_api/ with one module per domain (accounts, entries, payments, settings_reports, reconciliation, permissions, admin), each registering full literal paths on its own APIRouter; __init__ builds the combined libra_api_router so libra/__init__.py is untouched. Shared imports/helpers live in views_api/_shared.py; _extract_entry_id and _SYSTEM_LINK_PREFIXES move to beancount_format.py (they are pure entry-dict parsing). Route behavior is pinned by tests/test_route_table.py: the 73-route set is unchanged, and the two order-sensitive families (the shadowed /accounts/hierarchy wart — deliberately preserved here, fixed in the next commit — and the admin sync literal/param pair) keep their relative order inside a single module each. Addresses CODE-REVIEW-2026-06 structure findings (views_api monolith). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
9e06fa0b2a
commit
a7d7740a3a
12 changed files with 4233 additions and 4130 deletions
210
views_api/permissions.py
Normal file
210
views_api/permissions.py
Normal file
|
|
@ -0,0 +1,210 @@
|
|||
"""Libra API — permissions endpoints (moved verbatim from views_api.py)."""
|
||||
from fastapi import APIRouter
|
||||
|
||||
from ._shared import * # noqa: F401,F403
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.post("/api/v1/admin/equity-eligibility", status_code=HTTPStatus.CREATED)
|
||||
async def api_grant_equity_eligibility(
|
||||
data: CreateUserEquityStatus,
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> UserEquityStatus:
|
||||
"""Grant equity contribution eligibility to a user (admin only)"""
|
||||
from ..crud import create_or_update_user_equity_status
|
||||
|
||||
return await create_or_update_user_equity_status(data, auth.user_id)
|
||||
|
||||
|
||||
@router.delete("/api/v1/admin/equity-eligibility/{user_id}")
|
||||
async def api_revoke_equity_eligibility(
|
||||
user_id: str,
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> UserEquityStatus:
|
||||
"""Revoke equity contribution eligibility from a user (admin only)"""
|
||||
from ..crud import revoke_user_equity_eligibility
|
||||
|
||||
result = await revoke_user_equity_eligibility(user_id)
|
||||
if not result:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"User {user_id} not found in equity status records",
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@router.get("/api/v1/admin/equity-eligibility")
|
||||
async def api_list_equity_eligible_users(
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> list[UserEquityStatus]:
|
||||
"""List all equity-eligible users (admin only)"""
|
||||
from ..crud import get_all_equity_eligible_users
|
||||
|
||||
return await get_all_equity_eligible_users()
|
||||
|
||||
|
||||
# ===== ACCOUNT PERMISSION ADMIN ENDPOINTS =====
|
||||
|
||||
|
||||
@router.post("/api/v1/admin/permissions", status_code=HTTPStatus.CREATED)
|
||||
async def api_grant_permission(
|
||||
data: CreateAccountPermission,
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> AccountPermission:
|
||||
"""Grant account permission to a user (admin only)"""
|
||||
# Validate that account exists
|
||||
account = await get_account(data.account_id)
|
||||
if not account:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"Account with ID '{data.account_id}' not found",
|
||||
)
|
||||
|
||||
return await create_account_permission(data, auth.user_id)
|
||||
|
||||
|
||||
@router.get("/api/v1/admin/permissions")
|
||||
async def api_list_permissions(
|
||||
user_id: str | None = None,
|
||||
account_id: str | None = None,
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> list[AccountPermission]:
|
||||
"""
|
||||
List account permissions (admin only).
|
||||
Can filter by user_id or account_id.
|
||||
"""
|
||||
if user_id:
|
||||
return await get_user_permissions(user_id)
|
||||
elif account_id:
|
||||
return await get_account_permissions(account_id)
|
||||
else:
|
||||
# Get all permissions (get all users' permissions)
|
||||
# This is a bit inefficient but works for admin overview
|
||||
all_accounts = await get_all_accounts()
|
||||
all_permissions = []
|
||||
for account in all_accounts:
|
||||
account_perms = await get_account_permissions(account.id)
|
||||
all_permissions.extend(account_perms)
|
||||
|
||||
# Deduplicate by permission ID
|
||||
seen_ids = set()
|
||||
unique_permissions = []
|
||||
for perm in all_permissions:
|
||||
if perm.id not in seen_ids:
|
||||
seen_ids.add(perm.id)
|
||||
unique_permissions.append(perm)
|
||||
|
||||
return unique_permissions
|
||||
|
||||
|
||||
@router.delete("/api/v1/admin/permissions/{permission_id}")
|
||||
async def api_revoke_permission(
|
||||
permission_id: str,
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> dict:
|
||||
"""Revoke (delete) an account permission (admin only)"""
|
||||
# Verify permission exists
|
||||
permission = await get_account_permission(permission_id)
|
||||
if not permission:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"Permission with ID '{permission_id}' not found",
|
||||
)
|
||||
|
||||
await delete_account_permission(permission_id)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"message": f"Permission {permission_id} revoked successfully",
|
||||
}
|
||||
|
||||
|
||||
@router.post("/api/v1/admin/permissions/bulk", status_code=HTTPStatus.CREATED)
|
||||
async def api_bulk_grant_permissions(
|
||||
permissions: list[CreateAccountPermission],
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> list[AccountPermission]:
|
||||
"""Grant multiple account permissions at once (admin only)"""
|
||||
created_permissions = []
|
||||
|
||||
for perm_data in permissions:
|
||||
# Validate that account exists
|
||||
account = await get_account(perm_data.account_id)
|
||||
if not account:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"Account with ID '{perm_data.account_id}' not found",
|
||||
)
|
||||
|
||||
perm = await create_account_permission(perm_data, auth.user_id)
|
||||
created_permissions.append(perm)
|
||||
|
||||
return created_permissions
|
||||
|
||||
|
||||
@router.post("/api/v1/admin/permissions/bulk-grant", status_code=HTTPStatus.CREATED)
|
||||
async def api_bulk_grant_permission_to_users(
|
||||
data: "BulkGrantPermission",
|
||||
auth: AuthContext = Depends(require_super_user),
|
||||
) -> "BulkGrantResult":
|
||||
"""
|
||||
Grant the same permission to multiple users at once (admin only).
|
||||
|
||||
This is a convenience endpoint that grants the same account permission
|
||||
to multiple users in one operation. Useful for onboarding teams or
|
||||
granting access to a shared expense account.
|
||||
|
||||
Returns detailed results including successes and failures.
|
||||
"""
|
||||
from ..models import BulkGrantResult
|
||||
|
||||
granted = []
|
||||
failed = []
|
||||
|
||||
# Validate account exists and is active
|
||||
account = await get_account(data.account_id)
|
||||
if not account:
|
||||
raise HTTPException(
|
||||
status_code=HTTPStatus.NOT_FOUND,
|
||||
detail=f"Account with ID '{data.account_id}' not found",
|
||||
)
|
||||
|
||||
# Grant permission to each user
|
||||
for user_id in data.user_ids:
|
||||
try:
|
||||
perm_data = CreateAccountPermission(
|
||||
user_id=user_id,
|
||||
account_id=data.account_id,
|
||||
permission_type=data.permission_type,
|
||||
expires_at=data.expires_at,
|
||||
notes=data.notes,
|
||||
)
|
||||
perm = await create_account_permission(perm_data, auth.user_id)
|
||||
granted.append(perm)
|
||||
except Exception as e:
|
||||
failed.append({
|
||||
"user_id": user_id,
|
||||
"error": str(e),
|
||||
})
|
||||
|
||||
return BulkGrantResult(
|
||||
granted=granted,
|
||||
failed=failed,
|
||||
total=len(data.user_ids),
|
||||
success_count=len(granted),
|
||||
failure_count=len(failed),
|
||||
)
|
||||
|
||||
|
||||
# ===== USER PERMISSION ENDPOINTS =====
|
||||
|
||||
|
||||
@router.get("/api/v1/users/me/permissions")
|
||||
async def api_get_user_permissions(
|
||||
wallet: WalletTypeInfo = Depends(require_invoice_key),
|
||||
) -> list[AccountPermission]:
|
||||
"""Get current user's account permissions"""
|
||||
return await get_user_permissions(wallet.wallet.user)
|
||||
|
||||
|
||||
# ===== ACCOUNT HIERARCHY ENDPOINT =====
|
||||
Loading…
Add table
Add a link
Reference in a new issue