Jonathan Kirkwood
ae967494bd
The external fund_administrator role (relabeled Administrator) now signs
into the full admin interface, fenced to the funds and SPVs granted to
it via EntityAccess:
- Partners, capital accounts, documents (upload and delete), entity
edits, and eNAV imports for its own funds only; no fund creation,
valuation sign-off, audit log, or investor view.
- Scoped user management: sees and manages only investors tied to its
funds; creates investor accounts only; updates preserve grants on
funds outside its scope.
- New DELETE /api/users/{id} (in-app Delete user button) with the
cascade cleanup factored out of the CLI; Service Admin and self are
protected, and an Administrator can only delete an investor who
belongs solely to its funds.
- Internal fund_admin relabeled 'Staff (all funds)' and dropped from
the create picker to end the two-similar-names confusion.
- Version badge removed from the UI (sidebar and portal header); the
build version now logs to the browser console instead.
- deploy/.startos (signing key) added to .gitignore.
569 lines
23 KiB
Python
569 lines
23 KiB
Python
"""User administration: create and manage accounts and their entity access."""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlmodel import Session, select, col
|
|
|
|
from ten31portal.audit import record_audit
|
|
from ten31portal.auth import (
|
|
accessible_entity_ids, can_access_entity, check_administrator_scope, get_current_user,
|
|
hash_password, household_user_ids, require_admin, require_internal_admin,
|
|
)
|
|
from ten31portal.database import get_session
|
|
from ten31portal.models import (
|
|
AuditLog, CapitalAccountStatement, Document, Entity, EntityAccess, EXTERNAL_ROLES,
|
|
User, UserRole, ValuationRound,
|
|
)
|
|
from ten31portal.schemas import (
|
|
AccessGrant, AccessMatrixResponse, AccountLink, CapitalAccountResponse,
|
|
DocumentResponse, EntityResponse, InvestorViewResponse, LinkedAccount, PasswordReset,
|
|
UserCreate, UserDetailResponse, UserResponse, UserUpdate,
|
|
)
|
|
|
|
router = APIRouter(prefix="/api/users", tags=["users"])
|
|
|
|
|
|
# --- Administrator (external, entity-scoped) visibility helpers ---
|
|
|
|
def _visible_investor_ids(scope: set[int], session: Session) -> set[int]:
|
|
"""Investors an Administrator may see and manage: anyone with an access grant or a
|
|
capital-account statement in one of the Administrator's entities."""
|
|
if not scope:
|
|
return set()
|
|
ids = set(session.exec(
|
|
select(EntityAccess.user_id).where(col(EntityAccess.entity_id).in_(scope))
|
|
).all())
|
|
ids |= set(session.exec(
|
|
select(CapitalAccountStatement.investor_user_id).where(
|
|
col(CapitalAccountStatement.entity_id).in_(scope)
|
|
)
|
|
).all())
|
|
return ids
|
|
|
|
|
|
def _check_target_in_scope(admin: User, target: User, session: Session) -> None:
|
|
"""403 unless the caller is internal, or the target is an investor tied to one of the
|
|
Administrator's entities. Keeps an external Administrator away from staff accounts and
|
|
from other funds' investors entirely."""
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is None:
|
|
return
|
|
if target.role != UserRole.investor or target.id not in _visible_investor_ids(scope, session):
|
|
raise HTTPException(status_code=403, detail="No access to this account")
|
|
|
|
|
|
def delete_user_cascade(session: Session, user: User) -> list[str]:
|
|
"""Delete a user and its dependent rows; returns warnings for files that would not delete.
|
|
|
|
Shared by the API endpoint and the CLI. The caller is responsible for the guards
|
|
(Service Admin, self-deletion, scope) and for committing the session.
|
|
"""
|
|
from ten31portal import storage
|
|
|
|
warnings: list[str] = []
|
|
uid = user.id
|
|
|
|
# Entity-access grants and capital-account statements are this user's own data.
|
|
for acc in session.exec(select(EntityAccess).where(EntityAccess.user_id == uid)).all():
|
|
session.delete(acc)
|
|
for stmt in session.exec(
|
|
select(CapitalAccountStatement).where(CapitalAccountStatement.investor_user_id == uid)
|
|
).all():
|
|
session.delete(stmt)
|
|
|
|
# Documents addressed privately to this investor are removed (file + row); documents
|
|
# they uploaded stay, with the uploader cleared.
|
|
for doc in session.exec(select(Document).where(Document.investor_user_id == uid)).all():
|
|
try:
|
|
storage.delete_file(doc.storage_path)
|
|
except OSError as exc:
|
|
warnings.append(f"could not delete file {doc.storage_path}: {exc}")
|
|
session.delete(doc)
|
|
for doc in session.exec(select(Document).where(Document.uploaded_by == uid)).all():
|
|
doc.uploaded_by = None
|
|
session.add(doc)
|
|
|
|
# Preserve history/rounds by clearing the references to this user.
|
|
for rnd in session.exec(
|
|
select(ValuationRound).where(
|
|
(ValuationRound.submitted_by == uid) | (ValuationRound.approved_by == uid)
|
|
)
|
|
).all():
|
|
if rnd.submitted_by == uid:
|
|
rnd.submitted_by = None
|
|
if rnd.approved_by == uid:
|
|
rnd.approved_by = None
|
|
session.add(rnd)
|
|
for log in session.exec(select(AuditLog).where(AuditLog.actor_user_id == uid)).all():
|
|
log.actor_user_id = None
|
|
session.add(log)
|
|
|
|
# Detach any linked sub-accounts so they log in on their own again.
|
|
for sub in session.exec(select(User).where(User.primary_account_id == uid)).all():
|
|
sub.primary_account_id = None
|
|
session.add(sub)
|
|
|
|
session.delete(user)
|
|
return warnings
|
|
|
|
|
|
@router.get("/access-matrix")
|
|
def access_matrix(
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> AccessMatrixResponse:
|
|
"""External accounts, all entities, and the grants linking them.
|
|
|
|
For an external Administrator the matrix is their slice of the world: only their
|
|
granted entities, and only the investors tied to those entities.
|
|
"""
|
|
users = session.exec(
|
|
select(User).where(col(User.role).in_(EXTERNAL_ROLES)).order_by(User.name) # type: ignore[arg-type]
|
|
).all()
|
|
entities = session.exec(select(Entity).order_by(Entity.name)).all() # type: ignore[arg-type]
|
|
grants = session.exec(select(EntityAccess)).all()
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is not None:
|
|
visible = _visible_investor_ids(scope, session)
|
|
users = [u for u in users if u.id in visible]
|
|
entities = [e for e in entities if e.id in scope]
|
|
grants = [g for g in grants if g.entity_id in scope and g.user_id in visible]
|
|
return AccessMatrixResponse(
|
|
users=[UserResponse.model_validate(u, from_attributes=True) for u in users],
|
|
entities=[EntityResponse.model_validate(e, from_attributes=True) for e in entities],
|
|
grants=[AccessGrant(user_id=g.user_id, entity_id=g.entity_id) for g in grants],
|
|
)
|
|
|
|
|
|
@router.put("/{user_id}/access/{entity_id}", status_code=200)
|
|
def grant_access(
|
|
user_id: int,
|
|
entity_id: int,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, str]:
|
|
check_administrator_scope(admin, entity_id, session)
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if admin.role == UserRole.fund_administrator and user.role != UserRole.investor:
|
|
raise HTTPException(status_code=403, detail="Administrators manage investor accounts only")
|
|
if session.get(Entity, entity_id) is None:
|
|
raise HTTPException(status_code=404, detail="Entity not found")
|
|
existing = session.exec(
|
|
select(EntityAccess).where(
|
|
EntityAccess.user_id == user_id, EntityAccess.entity_id == entity_id
|
|
)
|
|
).first()
|
|
if existing is None:
|
|
session.add(EntityAccess(user_id=user_id, entity_id=entity_id))
|
|
record_audit(session, admin.id, "grant_access", "user", user_id, {"entity_id": entity_id})
|
|
session.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.delete("/{user_id}/access/{entity_id}")
|
|
def revoke_access(
|
|
user_id: int,
|
|
entity_id: int,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, str]:
|
|
check_administrator_scope(admin, entity_id, session)
|
|
if admin.role == UserRole.fund_administrator:
|
|
target = session.get(User, user_id)
|
|
if target is not None and target.role != UserRole.investor:
|
|
raise HTTPException(
|
|
status_code=403, detail="Administrators manage investor accounts only"
|
|
)
|
|
existing = session.exec(
|
|
select(EntityAccess).where(
|
|
EntityAccess.user_id == user_id, EntityAccess.entity_id == entity_id
|
|
)
|
|
).first()
|
|
if existing is not None:
|
|
session.delete(existing)
|
|
record_audit(session, admin.id, "revoke_access", "user", user_id, {"entity_id": entity_id})
|
|
session.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.get("/investors-for-entity/{entity_id}")
|
|
def investors_for_entity(
|
|
entity_id: int,
|
|
user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session),
|
|
) -> list[UserResponse]:
|
|
"""Investor accounts with access to an entity. For internal staff and the entity's fund admins."""
|
|
if user.role not in (UserRole.approver, UserRole.cfo, UserRole.operations) and not (
|
|
user.role == UserRole.fund_administrator
|
|
and can_access_entity(user, entity_id, session)
|
|
):
|
|
raise HTTPException(status_code=403, detail="Insufficient permissions")
|
|
rows = session.exec(
|
|
select(User)
|
|
.join(EntityAccess, EntityAccess.user_id == User.id)
|
|
.where(EntityAccess.entity_id == entity_id, User.role == UserRole.investor)
|
|
.order_by(User.name) # type: ignore[arg-type]
|
|
).all()
|
|
return [UserResponse.model_validate(r, from_attributes=True) for r in rows]
|
|
|
|
|
|
@router.get("/{user_id}/investor-view")
|
|
def investor_view(
|
|
user_id: int,
|
|
admin: User = Depends(require_internal_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> InvestorViewResponse:
|
|
"""Reconstruct exactly what an investor sees in their portal — read-only, for admins.
|
|
|
|
No session impersonation: this returns the same data the investor's own portal would load
|
|
(their accessible entities, their capital statements, and the documents visible to them),
|
|
scoped with the same access helpers.
|
|
"""
|
|
target = session.get(User, user_id)
|
|
if target is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if target.role != UserRole.investor:
|
|
raise HTTPException(status_code=400, detail="Investor View is for investor accounts.")
|
|
|
|
allowed = accessible_entity_ids(target, session) or set()
|
|
household = household_user_ids(target, session)
|
|
|
|
entities = session.exec(
|
|
select(Entity).where(col(Entity.id).in_(allowed)).order_by(Entity.name) # type: ignore[arg-type]
|
|
).all() if allowed else []
|
|
|
|
caps: list[CapitalAccountResponse] = []
|
|
docs: list[DocumentResponse] = []
|
|
if allowed:
|
|
cap_rows = session.exec(
|
|
select(CapitalAccountStatement)
|
|
.where(
|
|
col(CapitalAccountStatement.investor_user_id).in_(household),
|
|
col(CapitalAccountStatement.entity_id).in_(allowed),
|
|
)
|
|
.order_by(col(CapitalAccountStatement.as_of_date).desc())
|
|
).all()
|
|
names = dict(session.exec(
|
|
select(User.id, User.name).where(
|
|
col(User.id).in_({r.investor_user_id for r in cap_rows})
|
|
)
|
|
).all()) if cap_rows else {}
|
|
# Mirror the investor's own portal exactly — including exit status and BTC marks,
|
|
# so this view never drifts from what the LP actually sees.
|
|
from ten31portal.routers.capital_account_router import btc_marks, exit_dates
|
|
exits = exit_dates(session, cap_rows)
|
|
btc_asof, btc_close = btc_marks(session, cap_rows)
|
|
for r in cap_rows:
|
|
d = CapitalAccountResponse.model_validate(r, from_attributes=True)
|
|
d.investor_name = names.get(r.investor_user_id)
|
|
d.exited_on = exits.get((r.investor_user_id, r.entity_id))
|
|
d.btc_price_cents = btc_asof.get(r.id)
|
|
d.btc_close_price_cents = btc_close.get(r.entity_id)
|
|
caps.append(d)
|
|
|
|
doc_rows = session.exec(
|
|
select(Document)
|
|
.where(col(Document.entity_id).in_(allowed))
|
|
.order_by(col(Document.created_at).desc())
|
|
).all()
|
|
# Investor sees shared docs and those addressed to any of their linked names.
|
|
docs = [
|
|
DocumentResponse.model_validate(d, from_attributes=True)
|
|
for d in doc_rows
|
|
if d.investor_user_id is None or d.investor_user_id in household
|
|
]
|
|
|
|
return InvestorViewResponse(
|
|
user=UserResponse.model_validate(target, from_attributes=True),
|
|
entities=[EntityResponse.model_validate(e, from_attributes=True) for e in entities],
|
|
capital_accounts=caps,
|
|
documents=docs,
|
|
)
|
|
|
|
|
|
def _entity_ids_for(user_id: int, session: Session) -> list[int]:
|
|
return list(session.exec(
|
|
select(EntityAccess.entity_id).where(EntityAccess.user_id == user_id)
|
|
).all())
|
|
|
|
|
|
def _user_detail(user: User, session: Session) -> UserDetailResponse:
|
|
"""Build a full user detail, including the linked-account relationships."""
|
|
data = UserResponse.model_validate(user, from_attributes=True).model_dump()
|
|
primary_name = None
|
|
if user.primary_account_id:
|
|
primary = session.get(User, user.primary_account_id)
|
|
primary_name = primary.name if primary else None
|
|
linked = session.exec(
|
|
select(User).where(User.primary_account_id == user.id).order_by(User.name) # type: ignore[arg-type]
|
|
).all()
|
|
return UserDetailResponse(
|
|
**data,
|
|
primary_account_name=primary_name,
|
|
linked_accounts=[
|
|
LinkedAccount(id=u.id, name=u.name, username=u.username) for u in linked
|
|
],
|
|
entity_ids=_entity_ids_for(user.id, session),
|
|
)
|
|
|
|
|
|
def _set_entity_access(user_id: int, entity_ids: list[int], session: Session) -> None:
|
|
"""Replace a user's entity grants with the given set, ignoring unknown ids."""
|
|
valid = set(session.exec(
|
|
select(Entity.id).where(Entity.id.in_(entity_ids)) # type: ignore[union-attr]
|
|
).all()) if entity_ids else set()
|
|
existing = session.exec(
|
|
select(EntityAccess).where(EntityAccess.user_id == user_id)
|
|
).all()
|
|
current = {a.entity_id: a for a in existing}
|
|
# Remove grants no longer wanted.
|
|
for eid, access in current.items():
|
|
if eid not in valid:
|
|
session.delete(access)
|
|
# Add new grants.
|
|
for eid in valid:
|
|
if eid not in current:
|
|
session.add(EntityAccess(user_id=user_id, entity_id=eid))
|
|
|
|
|
|
@router.get("")
|
|
def list_users(
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> list[UserResponse]:
|
|
rows = session.exec(select(User).order_by(User.name)).all() # type: ignore[arg-type]
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is not None:
|
|
visible = _visible_investor_ids(scope, session)
|
|
rows = [r for r in rows if r.role == UserRole.investor and r.id in visible]
|
|
return [UserResponse.model_validate(r, from_attributes=True) for r in rows]
|
|
|
|
|
|
@router.get("/{user_id}")
|
|
def get_user(
|
|
user_id: int,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> UserDetailResponse:
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
_check_target_in_scope(admin, user, session)
|
|
return _user_detail(user, session)
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
def create_user(
|
|
body: UserCreate,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> UserDetailResponse:
|
|
entity_ids = body.entity_ids
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is not None:
|
|
# An Administrator creates investor accounts only, and only on their own funds —
|
|
# anything else would be privilege escalation or an account they can't see again.
|
|
if body.role != UserRole.investor:
|
|
raise HTTPException(
|
|
status_code=403, detail="Administrators can only create investor accounts"
|
|
)
|
|
entity_ids = [e for e in entity_ids if e in scope]
|
|
if not entity_ids:
|
|
raise HTTPException(
|
|
status_code=400, detail="Select at least one of your funds for this investor"
|
|
)
|
|
if session.exec(select(User).where(User.username == body.username)).first():
|
|
raise HTTPException(status_code=409, detail="Username already taken")
|
|
if body.email and session.exec(select(User).where(User.email == body.email)).first():
|
|
raise HTTPException(status_code=409, detail="Email already in use")
|
|
|
|
user = User(
|
|
name=body.name,
|
|
username=body.username,
|
|
email=body.email or None,
|
|
password_hash=hash_password(body.password),
|
|
role=body.role,
|
|
)
|
|
session.add(user)
|
|
session.flush()
|
|
_set_entity_access(user.id, entity_ids, session)
|
|
record_audit(session, admin.id, "create", "user", user.id,
|
|
{"username": body.username, "role": body.role.value,
|
|
"entity_ids": entity_ids})
|
|
session.commit()
|
|
session.refresh(user)
|
|
return _user_detail(user, session)
|
|
|
|
|
|
@router.patch("/{user_id}")
|
|
def update_user(
|
|
user_id: int,
|
|
body: UserUpdate,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> UserDetailResponse:
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
_check_target_in_scope(admin, user, session)
|
|
|
|
changes = body.model_dump(exclude_unset=True)
|
|
entity_ids = changes.pop("entity_ids", None)
|
|
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is not None:
|
|
if "role" in changes and changes["role"] != user.role:
|
|
raise HTTPException(status_code=403, detail="Administrators cannot change roles")
|
|
changes.pop("role", None)
|
|
if entity_ids is not None:
|
|
# Only this Administrator's funds are theirs to grant or revoke; the investor's
|
|
# access to any other fund is preserved untouched.
|
|
kept_elsewhere = set(_entity_ids_for(user_id, session)) - scope
|
|
entity_ids = list((set(entity_ids) & scope) | kept_elsewhere)
|
|
|
|
if "username" in changes:
|
|
new_username = (changes["username"] or "").strip()
|
|
if not new_username:
|
|
raise HTTPException(status_code=400, detail="Username cannot be blank")
|
|
clash = session.exec(select(User).where(User.username == new_username)).first()
|
|
if clash and clash.id != user_id:
|
|
raise HTTPException(status_code=409, detail="Username already taken")
|
|
changes["username"] = new_username
|
|
|
|
if "email" in changes and changes["email"]:
|
|
clash = session.exec(select(User).where(User.email == changes["email"])).first()
|
|
if clash and clash.id != user_id:
|
|
raise HTTPException(status_code=409, detail="Email already in use")
|
|
|
|
for key, val in changes.items():
|
|
setattr(user, key, val)
|
|
session.add(user)
|
|
session.flush()
|
|
|
|
if entity_ids is not None:
|
|
_set_entity_access(user_id, entity_ids, session)
|
|
|
|
record_audit(session, admin.id, "update", "user", user_id,
|
|
{**changes, **({"entity_ids": entity_ids} if entity_ids is not None else {})})
|
|
session.commit()
|
|
session.refresh(user)
|
|
return _user_detail(user, session)
|
|
|
|
|
|
@router.put("/{user_id}/primary-account")
|
|
def link_account(
|
|
user_id: int,
|
|
body: AccountLink,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> UserDetailResponse:
|
|
"""Link an investor account to a primary login (or detach it when null).
|
|
|
|
The primary becomes the single sign-on that sees every linked name's investments. The
|
|
linked account's own login is disabled so there is one set of credentials per person.
|
|
"""
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
_check_target_in_scope(admin, user, session)
|
|
if user.role != UserRole.investor:
|
|
raise HTTPException(status_code=400, detail="Only investor accounts can be linked.")
|
|
|
|
primary_id = body.primary_account_id
|
|
if primary_id is not None:
|
|
if primary_id == user_id:
|
|
raise HTTPException(status_code=400, detail="An account cannot link to itself.")
|
|
primary = session.get(User, primary_id)
|
|
if primary is None or primary.role != UserRole.investor:
|
|
raise HTTPException(status_code=400, detail="Primary must be an investor account.")
|
|
_check_target_in_scope(admin, primary, session)
|
|
if primary.primary_account_id is not None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="That account is itself linked to another login. Link to a primary instead.",
|
|
)
|
|
# Prevent chains: an account that other names log in under can't become a secondary.
|
|
if session.exec(select(User).where(User.primary_account_id == user_id)).first():
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="This account is a primary for other names. Detach those first.",
|
|
)
|
|
# Login is blocked while primary_account_id is set (see auth_router.login), so we leave
|
|
# login_enabled untouched — unlinking then restores the account's own sign-in cleanly.
|
|
user.primary_account_id = primary_id
|
|
else:
|
|
user.primary_account_id = None
|
|
|
|
session.add(user)
|
|
record_audit(session, admin.id, "link_account", "user", user_id,
|
|
{"primary_account_id": primary_id})
|
|
session.commit()
|
|
session.refresh(user)
|
|
return _user_detail(user, session)
|
|
|
|
|
|
@router.post("/{user_id}/reset-password")
|
|
def reset_password(
|
|
user_id: int,
|
|
body: PasswordReset,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, str]:
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
_check_target_in_scope(admin, user, session)
|
|
user.password_hash = hash_password(body.password)
|
|
# Admin handed them a real password — no forced change on next login.
|
|
user.must_change_password = False
|
|
user.login_enabled = True # setting a password enables login
|
|
session.add(user)
|
|
record_audit(session, admin.id, "reset_password", "user", user_id, None)
|
|
session.commit()
|
|
return {"status": "ok"}
|
|
|
|
|
|
@router.delete("/{user_id}")
|
|
def delete_user(
|
|
user_id: int,
|
|
admin: User = Depends(require_admin),
|
|
session: Session = Depends(get_session),
|
|
) -> dict[str, str]:
|
|
"""Delete an account and its dependent data (grants, statements, private documents).
|
|
|
|
The Service Admin can never be deleted, nor can you delete yourself. An external
|
|
Administrator may only delete an investor who belongs solely to their own funds —
|
|
an investor who also sits in another fund is another manager's problem too.
|
|
"""
|
|
user = session.get(User, user_id)
|
|
if user is None:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
if user.is_service_admin:
|
|
raise HTTPException(status_code=400, detail="The Service Admin cannot be deleted")
|
|
if user.id == admin.id:
|
|
raise HTTPException(status_code=400, detail="You cannot delete your own account")
|
|
_check_target_in_scope(admin, user, session)
|
|
|
|
scope = accessible_entity_ids(admin, session)
|
|
if scope is not None:
|
|
footprint = set(_entity_ids_for(user_id, session))
|
|
footprint |= set(session.exec(
|
|
select(CapitalAccountStatement.entity_id).where(
|
|
CapitalAccountStatement.investor_user_id == user_id
|
|
)
|
|
).all())
|
|
if not footprint <= scope:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail="This investor also belongs to funds outside your access; "
|
|
"remove them from your funds instead of deleting the account.",
|
|
)
|
|
|
|
name, username = user.name, user.username
|
|
delete_user_cascade(session, user)
|
|
record_audit(session, admin.id, "delete", "user", user_id,
|
|
{"username": username, "name": name})
|
|
session.commit()
|
|
return {"status": "deleted"}
|