payroll/accounts.py
Padreug 55f3dfac91 style: run black and ruff over the committed sources
Pure formatting — import ordering and black's line wrapping. No behaviour
change. Kept out of the feature commits so their diffs stay readable.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018jy52j9GRZ6XKa1Zt21LLj
2026-08-31 13:45:27 +02:00

67 lines
2.5 KiB
Python

"""Account directory — the read model behind the operator's user picker.
Payroll needs to answer one question core does not expose in a
payroll-shaped way: *which accounts exist, and which wallets could a
payout land in?* This module builds that list out of core's own crud, so
there is no second copy of the account table anywhere in this extension.
It reads accounts the operator does not own, which is exactly why every
caller of it is gated on `check_super_user` at the API boundary.
"""
from lnbits.core.crud import get_accounts, get_wallets
from lnbits.db import Filters
from .models import DirectoryUser, DirectoryWallet
# Core paginates accounts and defaults to 10 per page. A payroll picker
# wants the whole roster in one go; this cap keeps a pathological instance
# from building a five-figure dropdown, and the search box narrows anything
# larger server-side.
MAX_DIRECTORY_USERS = 500
async def list_directory_users(
search: str | None = None, limit: int = MAX_DIRECTORY_USERS
) -> list[DirectoryUser]:
"""Accounts with their (non-deleted) wallets, for the payout picker.
Accounts with no wallet are still returned: they are legitimate payroll
targets, they just need a wallet created before a contract can point at
one. Hiding them would make the operator wonder why a user they can see
in the admin UI is missing here.
"""
page = await get_accounts(
filters=Filters(search=search or None, limit=min(limit, MAX_DIRECTORY_USERS))
)
users: list[DirectoryUser] = []
for account in page.data:
wallets = await get_wallets(account.id)
users.append(
DirectoryUser(
id=account.id,
username=account.username,
email=account.email,
wallets=[
DirectoryWallet(id=w.id, name=w.name, balance_msat=w.balance_msat)
for w in wallets
],
)
)
# Readable ordering for a dropdown; core sorts by whatever the page
# query returned, which is account id.
users.sort(key=lambda u: (u.display_name or "").lower())
return users
async def owns_wallet(user_id: str, wallet_id: str) -> bool:
"""Whether `wallet_id` belongs to `user_id`.
The guard that stops a payroll contract from being pointed at a wallet
belonging to somebody other than the selected employee — the id pair
arrives from a form, and nothing else in the write path re-checks it.
"""
wallets = await get_wallets(user_id)
return any(w.id == wallet_id for w in wallets)