wizard firm creation

This commit is contained in:
A R R R Associates
2026-07-01 15:17:38 +05:30
parent a834dc0e9d
commit 779565ac6c
7 changed files with 923 additions and 0 deletions
+288
View File
@@ -0,0 +1,288 @@
from __future__ import annotations
from fastapi import APIRouter, Form, Request
from fastapi.responses import RedirectResponse
from sqlalchemy import select
from app.core.db.common import CommonSessionLocal
from app.core.http_responses import forbidden_response, ui_access_denied
from app.core.security.csrf import get_or_create_csrf_token, validate_csrf
from app.core.security.session_auth import get_current_user
from app.core.templating import templates
from app.modules.core.audit.service import model_snapshot, write_audit_log
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.tenancy.models import Branch, FinancialYear, Tenant
from app.modules.core.iam.models import User
from app.modules.wizards.service import (
FirmWizardError,
build_firm_wizard_payload,
create_firm_from_wizard,
default_ay_from_fy,
default_dates_from_fy,
validate_firm_wizard_payload,
)
router = APIRouter(prefix="/wizards", tags=["wizards-ui"])
def _ctx(request: Request, db, user, **extra):
ctx = {
"request": request,
"current_user": user,
"current_user_roles": get_user_roles(db, user.id),
"current_user_permissions": get_user_permissions(db, user.id),
"csrf_token": get_or_create_csrf_token(request),
}
ctx.update(extra)
return ctx
def _login_redirect():
return RedirectResponse(url="/login", status_code=303)
def _csrf_rejected(request: Request):
return forbidden_response(request, "CSRF validation failed")
def _is_system_admin(db, user) -> bool:
return "System Admin" in get_user_roles(db, user.id)
def _system_admin_or_denied(request: Request, db):
user = get_current_user(request, db=db)
if not user:
return None, _login_redirect()
if not _is_system_admin(db, user):
return None, ui_access_denied()
return user, None
def _default_form_values() -> dict:
start, end = default_dates_from_fy("2026-27")
return {
"tenant_code": "",
"tenant_name": "",
"firm_type": "partnership",
"default_timezone": "Asia/Kolkata",
"default_session_duration_minutes": 480,
"default_otp_required_roles_csv": "Partner,System Admin",
"default_storage_mode": "local_only",
"branch_code": "HO",
"branch_name": "Head Office",
"branch_timezone": "Asia/Kolkata",
"branch_address_line1": "",
"branch_address_line2": "",
"branch_city": "",
"branch_state": "",
"branch_pin_code": "",
"branch_gstin": "",
"branch_pan": "",
"admin_full_name": "",
"admin_email": "",
"admin_mobile": "",
"admin_designation": "Firm Admin",
"create_financial_year": True,
"fy_year_code": "2026-27",
"fy_assessment_year": default_ay_from_fy("2026-27"),
"fy_start_date": start.isoformat() if start else "",
"fy_end_date": end.isoformat() if end else "",
"fy_is_current": True,
}
@router.get("/system/firm/new")
def system_firm_new(request: Request):
db = CommonSessionLocal()
try:
user, denied = _system_admin_or_denied(request, db)
if denied:
return denied
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_new.html",
_ctx(
request,
db,
user,
title="Create Firm Wizard",
form=_default_form_values(),
errors=[],
),
)
finally:
db.close()
@router.post("/system/firm/preview")
def system_firm_preview(
request: Request,
tenant_code: str = Form(""),
tenant_name: str = Form(""),
firm_type: str = Form("partnership"),
default_timezone: str = Form("Asia/Kolkata"),
default_session_duration_minutes: int = Form(480),
default_otp_required_roles_csv: str = Form("Partner,System Admin"),
default_storage_mode: str = Form("local_only"),
branch_code: str = Form("HO"),
branch_name: str = Form("Head Office"),
branch_timezone: str = Form("Asia/Kolkata"),
branch_address_line1: str = Form(""),
branch_address_line2: str = Form(""),
branch_city: str = Form(""),
branch_state: str = Form(""),
branch_pin_code: str = Form(""),
branch_gstin: str = Form(""),
branch_pan: str = Form(""),
admin_full_name: str = Form(""),
admin_email: str = Form(""),
admin_mobile: str = Form(""),
admin_designation: str = Form("Firm Admin"),
create_financial_year: str | None = Form(None),
fy_year_code: str = Form(""),
fy_assessment_year: str = Form(""),
fy_start_date: str = Form(""),
fy_end_date: str = Form(""),
fy_is_current: str | None = Form(None),
csrf_token: str = Form(...),
):
try:
validate_csrf(request, csrf_token)
except PermissionError:
return _csrf_rejected(request)
db = CommonSessionLocal()
try:
user, denied = _system_admin_or_denied(request, db)
if denied:
return denied
form = build_firm_wizard_payload(locals())
errors = validate_firm_wizard_payload(db, form)
if errors:
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_new.html",
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=errors),
status_code=400,
)
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_preview.html",
_ctx(request, db, user, title="Preview Firm Creation", form=form, errors=[]),
)
finally:
db.close()
@router.post("/system/firm/confirm")
def system_firm_confirm(
request: Request,
tenant_code: str = Form(""),
tenant_name: str = Form(""),
firm_type: str = Form("partnership"),
default_timezone: str = Form("Asia/Kolkata"),
default_session_duration_minutes: int = Form(480),
default_otp_required_roles_csv: str = Form("Partner,System Admin"),
default_storage_mode: str = Form("local_only"),
branch_code: str = Form("HO"),
branch_name: str = Form("Head Office"),
branch_timezone: str = Form("Asia/Kolkata"),
branch_address_line1: str = Form(""),
branch_address_line2: str = Form(""),
branch_city: str = Form(""),
branch_state: str = Form(""),
branch_pin_code: str = Form(""),
branch_gstin: str = Form(""),
branch_pan: str = Form(""),
admin_full_name: str = Form(""),
admin_email: str = Form(""),
admin_mobile: str = Form(""),
admin_designation: str = Form("Firm Admin"),
create_financial_year: str | None = Form(None),
fy_year_code: str = Form(""),
fy_assessment_year: str = Form(""),
fy_start_date: str = Form(""),
fy_end_date: str = Form(""),
fy_is_current: str | None = Form(None),
csrf_token: str = Form(...),
):
try:
validate_csrf(request, csrf_token)
except PermissionError:
return _csrf_rejected(request)
db = CommonSessionLocal()
try:
user, denied = _system_admin_or_denied(request, db)
if denied:
return denied
form = build_firm_wizard_payload(locals())
errors = validate_firm_wizard_payload(db, form)
if errors:
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_new.html",
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=errors),
status_code=400,
)
try:
result = create_firm_from_wizard(db, form)
tenant_id = result.tenant.id
branch_id = result.branch.id
admin_id = result.firm_admin.id
fy_id = result.financial_year.id if result.financial_year else None
db.commit()
except FirmWizardError as exc:
db.rollback()
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_new.html",
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=[str(exc)]),
status_code=400,
)
except Exception as exc:
db.rollback()
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_new.html",
_ctx(request, db, user, title="Create Firm Wizard", form=form, errors=[f"Firm was not created: {exc}"]),
status_code=400,
)
tenant = db.execute(select(Tenant).where(Tenant.id == tenant_id)).scalar_one()
branch = db.execute(select(Branch).where(Branch.id == branch_id)).scalar_one()
firm_admin = db.execute(select(User).where(User.id == admin_id)).scalar_one()
financial_year = db.execute(select(FinancialYear).where(FinancialYear.id == fy_id)).scalar_one_or_none() if fy_id else None
write_audit_log(
db,
action="wizard.system.firm.create",
entity_type="tenant",
actor=user,
request=request,
entity_id=tenant.id,
entity_name=tenant.name,
target_tenant_id=tenant.id,
target_branch_id=branch.id,
details={
"tenant": model_snapshot(tenant, ["code", "name", "firm_type", "is_active", "default_timezone", "default_storage_mode"]),
"branch": model_snapshot(branch, ["code", "name", "is_head_office", "allow_login", "allow_new_assignments"]),
"firm_admin": model_snapshot(firm_admin, ["email", "full_name", "tenant_id", "branch_id", "is_active", "allow_login", "must_change_password"]),
"financial_year": model_snapshot(financial_year, ["year_code", "assessment_year", "start_date", "end_date", "is_current"]) if financial_year else None,
},
)
return templates.TemplateResponse(
"modules/wizards/templates/wizards/system_firm_complete.html",
_ctx(
request,
db,
user,
title="Firm Created",
tenant=tenant,
branch=branch,
firm_admin=firm_admin,
financial_year=financial_year,
invite_url=result.invite_url,
),
)
finally:
db.close()