Add edit actions for business branch and registration
This commit is contained in:
A R R R Associates
@@ -3,6 +3,7 @@ from __future__ import annotations
|
||||
from fastapi import APIRouter, Form, Request
|
||||
from fastapi.responses import RedirectResponse
|
||||
from sqlalchemy import select
|
||||
from urllib.parse import urlencode
|
||||
|
||||
from app.core.db.common import CommonSessionLocal
|
||||
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
|
||||
@@ -43,6 +44,51 @@ def _load_client(db, tenant_id: int, client_id: int):
|
||||
return db.execute(select(Client).where(Client.id == client_id, Client.tenant_id == tenant_id)).scalar_one_or_none()
|
||||
|
||||
|
||||
|
||||
|
||||
def _structure_redirect(client_id: int, *, message: str | None = None, error: str | None = None, anchor: str | None = None):
|
||||
params = {}
|
||||
if message:
|
||||
params["message"] = message
|
||||
if error:
|
||||
params["error"] = error
|
||||
url = f"/clients/{client_id}/business-structure"
|
||||
if params:
|
||||
url += "?" + urlencode(params)
|
||||
if anchor:
|
||||
url += f"#{anchor}"
|
||||
return RedirectResponse(url, 303)
|
||||
|
||||
|
||||
def _owned_business(db, tenant_id: int, client_id: int, business_unit_id: int):
|
||||
return db.execute(
|
||||
select(ClientBusinessUnit).where(
|
||||
ClientBusinessUnit.id == business_unit_id,
|
||||
ClientBusinessUnit.tenant_id == tenant_id,
|
||||
ClientBusinessUnit.client_id == client_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def _owned_branch(db, tenant_id: int, client_id: int, branch_id: int):
|
||||
return db.execute(
|
||||
select(ClientBranch).where(
|
||||
ClientBranch.id == branch_id,
|
||||
ClientBranch.tenant_id == tenant_id,
|
||||
ClientBranch.client_id == client_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
|
||||
def _owned_registration(db, tenant_id: int, client_id: int, registration_id: int):
|
||||
return db.execute(
|
||||
select(ClientRegistration).where(
|
||||
ClientRegistration.id == registration_id,
|
||||
ClientRegistration.tenant_id == tenant_id,
|
||||
ClientRegistration.client_id == client_id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
def _context(request, db, user, **extra):
|
||||
data = {
|
||||
"request": request, "current_user": user,
|
||||
@@ -81,9 +127,7 @@ def business_structure_page(request: Request, client_id: int):
|
||||
registrations = [row[0] for row in registration_rows]
|
||||
registration_type_codes = {row[0].id: row[1].code for row in registration_rows}
|
||||
registration_types = db.execute(
|
||||
select(RegistrationType).where(RegistrationType.is_active.is_(True)).order_by(
|
||||
RegistrationType.sort_order, RegistrationType.name
|
||||
)
|
||||
select(RegistrationType).order_by(RegistrationType.sort_order, RegistrationType.name)
|
||||
).scalars().all()
|
||||
|
||||
# Credential Vault stays the single source of truth. This page only surfaces
|
||||
@@ -233,6 +277,209 @@ def add_registration(request: Request, client_id: int, business_unit_id: str = F
|
||||
db.close()
|
||||
|
||||
|
||||
@router.post("/{client_id}/business-units/{business_unit_id}/edit")
|
||||
def edit_business_unit(
|
||||
request: Request,
|
||||
client_id: int,
|
||||
business_unit_id: int,
|
||||
business_code: str = Form(...),
|
||||
business_name: str = Form(...),
|
||||
trade_name: str = Form(""),
|
||||
nature_of_business: str = Form(""),
|
||||
is_primary: str | None = Form(None),
|
||||
csrf_token: str = Form(...),
|
||||
):
|
||||
validate_csrf(request, csrf_token)
|
||||
db = CommonSessionLocal()
|
||||
try:
|
||||
user = get_current_user(request, db=db)
|
||||
require_permission(db, user, "clients.edit")
|
||||
tenant_id = _tenant_id(request, user)
|
||||
client = _load_client(db, tenant_id, client_id)
|
||||
row = _owned_business(db, tenant_id, client_id, business_unit_id) if client else None
|
||||
if not client or not row:
|
||||
return RedirectResponse("/clients", 303)
|
||||
|
||||
code = business_code.strip().upper()
|
||||
name = business_name.strip()
|
||||
if not code or not name:
|
||||
return _structure_redirect(client_id, error="Business code and business unit name are required.", anchor="business-units")
|
||||
|
||||
duplicate = db.execute(
|
||||
select(ClientBusinessUnit.id).where(
|
||||
ClientBusinessUnit.tenant_id == tenant_id,
|
||||
ClientBusinessUnit.client_id == client_id,
|
||||
ClientBusinessUnit.business_code == code,
|
||||
ClientBusinessUnit.id != row.id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if duplicate:
|
||||
return _structure_redirect(client_id, error=f"Business code {code} is already used for this client.", anchor="business-units")
|
||||
|
||||
if is_primary:
|
||||
db.query(ClientBusinessUnit).filter(
|
||||
ClientBusinessUnit.tenant_id == tenant_id,
|
||||
ClientBusinessUnit.client_id == client_id,
|
||||
).update({"is_primary": False}, synchronize_session=False)
|
||||
|
||||
row.business_code = code
|
||||
row.business_name = name
|
||||
row.trade_name = trade_name.strip() or None
|
||||
row.nature_of_business = nature_of_business.strip() or None
|
||||
row.is_primary = bool(is_primary)
|
||||
row.updated_by_user_id = user.id
|
||||
db.commit()
|
||||
return _structure_redirect(client_id, message="Business Unit updated successfully.", anchor="business-units")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@router.post("/{client_id}/branches/{client_branch_id}/edit")
|
||||
def edit_client_branch(
|
||||
request: Request,
|
||||
client_id: int,
|
||||
client_branch_id: int,
|
||||
business_unit_id: int = Form(...),
|
||||
branch_code: str = Form(...),
|
||||
branch_name: str = Form(...),
|
||||
branch_type: str = Form("branch"),
|
||||
city: str = Form(""),
|
||||
state: str = Form(""),
|
||||
pincode: str = Form(""),
|
||||
is_primary: str | None = Form(None),
|
||||
csrf_token: str = Form(...),
|
||||
):
|
||||
validate_csrf(request, csrf_token)
|
||||
db = CommonSessionLocal()
|
||||
try:
|
||||
user = get_current_user(request, db=db)
|
||||
require_permission(db, user, "clients.edit")
|
||||
tenant_id = _tenant_id(request, user)
|
||||
client = _load_client(db, tenant_id, client_id)
|
||||
row = _owned_branch(db, tenant_id, client_id, client_branch_id) if client else None
|
||||
business = _owned_business(db, tenant_id, client_id, business_unit_id) if client else None
|
||||
if not client or not row or not business:
|
||||
return RedirectResponse("/clients", 303)
|
||||
|
||||
code = branch_code.strip().upper()
|
||||
name = branch_name.strip()
|
||||
if not code or not name:
|
||||
return _structure_redirect(client_id, error="Branch code and branch name are required.", anchor="client-branches")
|
||||
|
||||
duplicate = db.execute(
|
||||
select(ClientBranch.id).where(
|
||||
ClientBranch.tenant_id == tenant_id,
|
||||
ClientBranch.business_unit_id == business.id,
|
||||
ClientBranch.branch_code == code,
|
||||
ClientBranch.id != row.id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if duplicate:
|
||||
return _structure_redirect(client_id, error=f"Branch code {code} is already used under the selected Business Unit.", anchor="client-branches")
|
||||
|
||||
if is_primary:
|
||||
db.query(ClientBranch).filter(
|
||||
ClientBranch.tenant_id == tenant_id,
|
||||
ClientBranch.business_unit_id == business.id,
|
||||
).update({"is_primary": False}, synchronize_session=False)
|
||||
|
||||
row.business_unit_id = business.id
|
||||
row.branch_code = code
|
||||
row.branch_name = name
|
||||
row.branch_type = branch_type.strip() or "branch"
|
||||
row.city = city.strip() or None
|
||||
row.state = state.strip() or None
|
||||
row.pincode = pincode.strip() or None
|
||||
row.is_primary = bool(is_primary)
|
||||
row.updated_by_user_id = user.id
|
||||
db.commit()
|
||||
return _structure_redirect(client_id, message="Client Branch updated successfully.", anchor="client-branches")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@router.post("/{client_id}/registrations/{registration_id}/edit")
|
||||
def edit_registration(
|
||||
request: Request,
|
||||
client_id: int,
|
||||
registration_id: int,
|
||||
business_unit_id: str = Form(""),
|
||||
client_branch_id: str = Form(""),
|
||||
registration_type_id: int = Form(...),
|
||||
registration_number: str = Form(...),
|
||||
legal_name: str = Form(""),
|
||||
trade_name: str = Form(""),
|
||||
state: str = Form(""),
|
||||
csrf_token: str = Form(...),
|
||||
):
|
||||
validate_csrf(request, csrf_token)
|
||||
db = CommonSessionLocal()
|
||||
try:
|
||||
user = get_current_user(request, db=db)
|
||||
require_permission(db, user, "clients.edit")
|
||||
tenant_id = _tenant_id(request, user)
|
||||
client = _load_client(db, tenant_id, client_id)
|
||||
row = _owned_registration(db, tenant_id, client_id, registration_id) if client else None
|
||||
if not client or not row:
|
||||
return RedirectResponse("/clients", 303)
|
||||
|
||||
try:
|
||||
bu_id = int(business_unit_id) if business_unit_id else None
|
||||
br_id = int(client_branch_id) if client_branch_id else None
|
||||
except (TypeError, ValueError):
|
||||
return _structure_redirect(client_id, error="Invalid Business Unit or Client Branch selection.", anchor="registrations")
|
||||
|
||||
business = _owned_business(db, tenant_id, client_id, bu_id) if bu_id else None
|
||||
if bu_id and not business:
|
||||
return _structure_redirect(client_id, error="Selected Business Unit does not belong to this client.", anchor="registrations")
|
||||
|
||||
branch = _owned_branch(db, tenant_id, client_id, br_id) if br_id else None
|
||||
if br_id and not branch:
|
||||
return _structure_redirect(client_id, error="Selected Client Branch does not belong to this client.", anchor="registrations")
|
||||
if branch:
|
||||
bu_id = branch.business_unit_id
|
||||
|
||||
registration_type = db.get(RegistrationType, registration_type_id)
|
||||
if not registration_type:
|
||||
return _structure_redirect(client_id, error="Registration Type was not found.", anchor="registrations")
|
||||
if not registration_type.is_active and registration_type.id != row.registration_type_id:
|
||||
return _structure_redirect(client_id, error="An inactive Registration Type cannot be newly selected.", anchor="registrations")
|
||||
|
||||
number = registration_number.strip().upper()
|
||||
if not number:
|
||||
return _structure_redirect(client_id, error="Registration number is required.", anchor="registrations")
|
||||
|
||||
duplicate = db.execute(
|
||||
select(ClientRegistration.id).where(
|
||||
ClientRegistration.tenant_id == tenant_id,
|
||||
ClientRegistration.registration_type_id == registration_type.id,
|
||||
ClientRegistration.registration_number == number,
|
||||
ClientRegistration.id != row.id,
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
if duplicate:
|
||||
return _structure_redirect(
|
||||
client_id,
|
||||
error=f"{registration_type.code} registration number {number} already exists.",
|
||||
anchor="registrations",
|
||||
)
|
||||
|
||||
old_state = row.state
|
||||
row.business_unit_id = bu_id
|
||||
row.client_branch_id = br_id
|
||||
row.registration_type_id = registration_type.id
|
||||
row.registration_number = number
|
||||
row.legal_name = legal_name.strip() or client.client_name
|
||||
row.trade_name = trade_name.strip() or None
|
||||
row.state = state.strip() or None
|
||||
if not row.jurisdiction or row.jurisdiction == old_state:
|
||||
row.jurisdiction = row.state
|
||||
db.commit()
|
||||
return _structure_redirect(client_id, message="Registration updated successfully.", anchor="registrations")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
@router.post("/{client_id}/business-structure/{entity}/{entity_id}/toggle")
|
||||
def toggle_scope_record(request: Request, client_id: int, entity: str, entity_id: int, csrf_token: str = Form(...)):
|
||||
validate_csrf(request, csrf_token)
|
||||
|
||||
Reference in New Issue
Block a user