Rakseeibra/arrr-erp
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
arrr-erp/app/modules/registrations/ui.py
T
A R R R Associates 4248aae4d3 Add bulk client registration addition
2026-08-06 14:32:59 +05:30

465 lines
28 KiB
Python
Raw Permalink Blame History

from __future__ import annotations
from datetime import date
from fastapi import APIRouter, Form, Request
from fastapi.responses import HTMLResponse, RedirectResponse
from sqlalchemy import or_, select
from sqlalchemy.exc import IntegrityError
from app.core.db.common import CommonSessionLocal
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.clients.models import Client, ClientBusinessUnit, ClientBranch
from app.modules.consultants.models import ConsultantProfile
from app.modules.core.rbac.deps import get_user_permissions, get_user_roles
from app.modules.core.rbac.permission_guard import require_permission
from app.modules.registrations.models import *
from app.modules.registrations.service import seed_registration_types, run_registration_lifecycle_once, complete_renewal
router=APIRouter(prefix="/registrations", tags=["registrations-ui"])
def _ctx(request,user,db,**kw):
base={"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)};base.update(kw);return base
def _user(request,db):
user=get_current_user(request,db)
if not user: return None
require_permission(db,user,"clients.view");return user
def _tenant(request,user): return int(request.session.get("active_tenant_id") or user.tenant_id)
def _branch(request,user):
raw=request.session.get("active_branch_id");return int(raw) if raw not in (None,"",0,"0") else None
def _can_manage(db,user):
try: require_permission(db,user,"clients.edit");return True
except Exception:return False
def _d(value): return date.fromisoformat(value) if value else None
@router.get("",response_class=HTMLResponse)
def dashboard(request:Request):
with CommonSessionLocal() as db:
user=_user(request,db)
if not user:return RedirectResponse('/login',303)
seed_registration_types(db);run_registration_lifecycle_once(db)
tid=_tenant(request,user); bid=_branch(request,user)
regq=select(ClientRegistration,Client,RegistrationType).join(Client,Client.id==ClientRegistration.client_id).join(RegistrationType,RegistrationType.id==ClientRegistration.registration_type_id).where(ClientRegistration.tenant_id==tid)
taskq=select(RegistrationRenewalTask,Client).join(Client,Client.id==RegistrationRenewalTask.client_id).where(RegistrationRenewalTask.tenant_id==tid,RegistrationRenewalTask.status.notin_(["completed","cancelled"]))
if bid: regq=regq.where(ClientRegistration.branch_id==bid);taskq=taskq.where(RegistrationRenewalTask.branch_id==bid)
regs=db.execute(regq.order_by(ClientRegistration.valid_until.asc().nullslast(),Client.client_name)).all();tasks=db.execute(taskq.order_by(RegistrationRenewalTask.cycle_due_date)).all()
counts={"active":0,"due":0,"expired":0,"tasks":len(tasks)}
for reg,_,_ in regs:
if reg.status=="expired":counts["expired"]+=1
elif reg.status in {"renewal_due","update_due"}:counts["due"]+=1
else:counts["active"]+=1
return templates.TemplateResponse("modules/registrations/templates/registrations/dashboard.html",_ctx(request,user,db,registrations=regs,tasks=tasks,counts=counts,can_manage=_can_manage(db,user),today=date.today()))
@router.get("/masters",response_class=HTMLResponse)
def masters(request:Request):
with CommonSessionLocal() as db:
user=_user(request,db)
if not user:return RedirectResponse('/login',303)
seed_registration_types(db);tid=_tenant(request,user)
return templates.TemplateResponse("modules/registrations/templates/registrations/masters.html",_ctx(request,user,db,types=db.execute(select(RegistrationType).order_by(RegistrationType.sort_order)).scalars().all(),rules=db.execute(select(RegistrationLifecycleRule,RegistrationType).join(RegistrationType).where(or_(RegistrationLifecycleRule.tenant_id.is_(None),RegistrationLifecycleRule.tenant_id==tid))).all(),templates_rows=db.execute(select(RegistrationDocumentTemplate,RegistrationType).join(RegistrationType).where(or_(RegistrationDocumentTemplate.tenant_id.is_(None),RegistrationDocumentTemplate.tenant_id==tid))).all(),can_manage=_can_manage(db,user)))
@router.post("/masters/types")
async def create_type(request:Request,code:str=Form(...),name:str=Form(...),category:str=Form("registration"),issuing_authority:str=Form(""),supports_expiry:bool=Form(False),supports_related_person:bool=Form(False),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit")
code=code.strip().upper();existing=db.execute(select(RegistrationType).where(RegistrationType.code==code)).scalar_one_or_none()
if existing: existing.name=name.strip();existing.category=category;existing.issuing_authority=issuing_authority or None;existing.supports_expiry=supports_expiry;existing.supports_related_person=supports_related_person;existing.is_active=True
else: db.add(RegistrationType(code=code,name=name.strip(),category=category,issuing_authority=issuing_authority or None,supports_expiry=supports_expiry,supports_related_person=supports_related_person,is_system=False))
db.commit();return RedirectResponse('/registrations/masters',303)
@router.post("/masters/rules")
async def create_rule(request:Request,registration_type_id:int=Form(...),entity_type:str=Form(""),cycle_type:str=Form(...),frequency_months:str=Form(""),renewal_window_days:int=Form(60),reminder_days_csv:str=Form("90,60,30,15,7,0"),grace_period_days:int=Form(0),requires_client_approval:bool=Form(False),requires_partner_review:bool=Form(False),auto_create_task:bool=Form(False),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit")
db.add(RegistrationLifecycleRule(tenant_id=_tenant(request,user),registration_type_id=registration_type_id,entity_type=entity_type or None,cycle_type=cycle_type,frequency_months=int(frequency_months) if frequency_months else None,renewal_window_days=renewal_window_days,reminder_days_csv=reminder_days_csv,grace_period_days=grace_period_days,requires_client_approval=requires_client_approval,requires_partner_review=requires_partner_review,auto_create_task=auto_create_task,created_by_user_id=user.id));db.commit();return RedirectResponse('/registrations/masters',303)
@router.post("/masters/documents")
async def create_doc_template(request:Request,registration_type_id:int=Form(...),action_type:str=Form(...),document_name:str=Form(...),description:str=Form(""),mandatory:bool=Form(False),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit")
db.add(RegistrationDocumentTemplate(tenant_id=_tenant(request,user),registration_type_id=registration_type_id,action_type=action_type,document_name=document_name.strip(),description=description or None,mandatory=mandatory));db.commit();return RedirectResponse('/registrations/masters',303)
@router.get("/bulk", response_class=HTMLResponse)
def bulk_registration_page(
request: Request,
created: int = 0,
skipped: int = 0,
invalid: int = 0,
):
with CommonSessionLocal() as db:
user = _user(request, db)
if not user:
return RedirectResponse("/login", 303)
require_permission(db, user, "clients.edit")
seed_registration_types(db)
tenant_id = _tenant(request, user)
branch_id = _branch(request, user)
clients_query = select(Client).where(
Client.tenant_id == tenant_id,
Client.is_active.is_(True),
)
if branch_id:
clients_query = clients_query.where(Client.branch_id == branch_id)
clients = db.execute(
clients_query.order_by(Client.client_name, Client.client_code)
).scalars().all()
client_ids = [row.id for row in clients]
business_units = []
client_branches = []
if client_ids:
business_units = db.execute(
select(ClientBusinessUnit).where(
ClientBusinessUnit.tenant_id == tenant_id,
ClientBusinessUnit.client_id.in_(client_ids),
ClientBusinessUnit.is_active.is_(True),
).order_by(
ClientBusinessUnit.client_id,
ClientBusinessUnit.is_primary.desc(),
ClientBusinessUnit.business_name,
)
).scalars().all()
client_branches = db.execute(
select(ClientBranch).where(
ClientBranch.tenant_id == tenant_id,
ClientBranch.client_id.in_(client_ids),
ClientBranch.is_active.is_(True),
).order_by(
ClientBranch.client_id,
ClientBranch.is_primary.desc(),
ClientBranch.branch_name,
)
).scalars().all()
businesses_by_client: dict[int, list[ClientBusinessUnit]] = {}
for row in business_units:
businesses_by_client.setdefault(row.client_id, []).append(row)
branches_by_client: dict[int, list[ClientBranch]] = {}
for row in client_branches:
branches_by_client.setdefault(row.client_id, []).append(row)
types = db.execute(
select(RegistrationType).where(
RegistrationType.is_active.is_(True)
).order_by(RegistrationType.sort_order, RegistrationType.name)
).scalars().all()
rules = db.execute(
select(RegistrationLifecycleRule, RegistrationType).join(
RegistrationType,
RegistrationType.id == RegistrationLifecycleRule.registration_type_id,
).where(
or_(
RegistrationLifecycleRule.tenant_id.is_(None),
RegistrationLifecycleRule.tenant_id == tenant_id,
),
RegistrationLifecycleRule.is_active.is_(True),
).order_by(RegistrationType.name, RegistrationLifecycleRule.id)
).all()
consultants = db.execute(
select(ConsultantProfile).where(
ConsultantProfile.tenant_id == tenant_id,
ConsultantProfile.is_active.is_(True),
)
).scalars().all()
return templates.TemplateResponse(
"modules/registrations/templates/registrations/bulk.html",
_ctx(
request,
user,
db,
title="Bulk Registration Addition",
clients=clients,
businesses_by_client=businesses_by_client,
branches_by_client=branches_by_client,
types=types,
rules=rules,
consultants=consultants,
created=created,
skipped=skipped,
invalid=invalid,
),
)
@router.post("/bulk")
async def bulk_registration_submit(
request: Request,
client_ids: list[int] = Form([]),
registration_type_id: int = Form(...),
default_scope: str = Form("client"),
jurisdiction: str = Form(""),
state_code: str = Form(""),
issue_date: str = Form(""),
effective_from: str = Form(""),
valid_until: str = Form(""),
next_action_date: str = Form(""),
lifecycle_rule_id: str = Form(""),
responsible_party: str = Form("firm"),
assigned_consultant_id: str = Form(""),
auto_create_task: bool = Form(False),
notes: str = Form(""),
csrf_token: str = Form(...),
):
validate_csrf(request, csrf_token)
with CommonSessionLocal() as db:
user = _user(request, db)
if not user:
return RedirectResponse("/login", 303)
require_permission(db, user, "clients.edit")
tenant_id = _tenant(request, user)
branch_id = _branch(request, user)
registration_type = db.get(RegistrationType, registration_type_id)
if not registration_type or not registration_type.is_active:
return RedirectResponse("/registrations/bulk?invalid=1", 303)
selected_ids = list(dict.fromkeys(int(value) for value in client_ids))
if not selected_ids:
return RedirectResponse("/registrations/bulk?invalid=1", 303)
clients_query = select(Client).where(
Client.tenant_id == tenant_id,
Client.id.in_(selected_ids),
Client.is_active.is_(True),
)
if branch_id:
clients_query = clients_query.where(Client.branch_id == branch_id)
permitted_clients = db.execute(clients_query).scalars().all()
clients_by_id = {row.id: row for row in permitted_clients}
form = await request.form()
created = 0
skipped = 0
invalid = 0
for client_id in selected_ids:
client = clients_by_id.get(client_id)
if not client:
invalid += 1
continue
registration_number = (
str(form.get(f"registration_number_{client_id}", ""))
.strip()
.upper()
)
if not registration_number:
invalid += 1
continue
business_unit_id = None
client_branch_id = None
raw_business_id = str(
form.get(f"business_unit_id_{client_id}", "")
).strip()
raw_branch_id = str(
form.get(f"client_branch_id_{client_id}", "")
).strip()
if raw_business_id:
business = db.get(ClientBusinessUnit, int(raw_business_id))
if (
not business
or business.tenant_id != tenant_id
or business.client_id != client.id
or not business.is_active
):
invalid += 1
continue
business_unit_id = business.id
if raw_branch_id:
client_branch = db.get(ClientBranch, int(raw_branch_id))
if (
not client_branch
or client_branch.tenant_id != tenant_id
or client_branch.client_id != client.id
or not client_branch.is_active
):
invalid += 1
continue
client_branch_id = client_branch.id
business_unit_id = client_branch.business_unit_id
if default_scope == "primary_business" and not business_unit_id:
business = db.execute(
select(ClientBusinessUnit).where(
ClientBusinessUnit.tenant_id == tenant_id,
ClientBusinessUnit.client_id == client.id,
ClientBusinessUnit.is_active.is_(True),
).order_by(
ClientBusinessUnit.is_primary.desc(),
ClientBusinessUnit.id,
)
).scalars().first()
if not business:
invalid += 1
continue
business_unit_id = business.id
if default_scope == "primary_branch" and not client_branch_id:
client_branch = db.execute(
select(ClientBranch).where(
ClientBranch.tenant_id == tenant_id,
ClientBranch.client_id == client.id,
ClientBranch.is_active.is_(True),
).order_by(
ClientBranch.is_primary.desc(),
ClientBranch.id,
)
).scalars().first()
if not client_branch:
invalid += 1
continue
client_branch_id = client_branch.id
business_unit_id = client_branch.business_unit_id
existing = db.execute(
select(ClientRegistration).where(
ClientRegistration.tenant_id == tenant_id,
ClientRegistration.registration_type_id == registration_type.id,
ClientRegistration.registration_number == registration_number,
)
).scalar_one_or_none()
if existing:
skipped += 1
continue
legal_name = str(form.get(f"legal_name_{client_id}", "")).strip()
trade_name = str(form.get(f"trade_name_{client_id}", "")).strip()
row_state = str(form.get(f"state_{client_id}", "")).strip()
db.add(
ClientRegistration(
tenant_id=tenant_id,
branch_id=client.branch_id,
client_id=client.id,
business_unit_id=business_unit_id,
client_branch_id=client_branch_id,
registration_type_id=registration_type.id,
registration_number=registration_number,
legal_name=legal_name or client.client_name,
trade_name=trade_name or client.trade_name or None,
state=row_state or client.state or None,
jurisdiction=jurisdiction.strip() or row_state or client.state or None,
state_code=state_code.strip().upper() or None,
issue_date=_d(issue_date),
effective_from=_d(effective_from),
valid_until=_d(valid_until),
next_action_date=_d(next_action_date),
lifecycle_rule_id=int(lifecycle_rule_id) if lifecycle_rule_id else None,
responsible_party=responsible_party,
assigned_user_id=client.partner_id,
assigned_consultant_id=(
int(assigned_consultant_id)
if assigned_consultant_id
else None
),
auto_create_task=auto_create_task,
status="active",
notes=notes.strip() or None,
created_by_user_id=user.id,
)
)
try:
db.flush()
except IntegrityError:
db.rollback()
skipped += 1
continue
created += 1
db.commit()
run_registration_lifecycle_once(db)
return RedirectResponse(
f"/registrations/bulk?created={created}&skipped={skipped}&invalid={invalid}",
status_code=303,
)
@router.get("/clients/{client_id}",response_class=HTMLResponse)
def client_register(request:Request,client_id:int):
with CommonSessionLocal() as db:
user=_user(request,db)
if not user:return RedirectResponse('/login',303)
client=db.get(Client,client_id)
if not client or client.tenant_id!=_tenant(request,user):return RedirectResponse('/registrations',303)
seed_registration_types(db)
regs=db.execute(select(ClientRegistration,RegistrationType).join(RegistrationType).where(ClientRegistration.client_id==client_id).order_by(RegistrationType.name)).all()
people=db.execute(select(ClientRelatedPerson).where(ClientRelatedPerson.client_id==client_id).order_by(ClientRelatedPerson.full_name)).scalars().all()
dscs=db.execute(select(ClientDigitalSignature,ClientRelatedPerson).join(ClientRelatedPerson).where(ClientDigitalSignature.client_id==client_id).order_by(ClientDigitalSignature.expires_on)).all()
rules=db.execute(select(RegistrationLifecycleRule).where(or_(RegistrationLifecycleRule.tenant_id.is_(None),RegistrationLifecycleRule.tenant_id==client.tenant_id),RegistrationLifecycleRule.is_active.is_(True))).scalars().all()
consultants=db.execute(select(ConsultantProfile).where(ConsultantProfile.tenant_id==client.tenant_id,ConsultantProfile.is_active.is_(True))).scalars().all()
return templates.TemplateResponse("modules/registrations/templates/registrations/client.html",_ctx(request,user,db,client=client,registrations=regs,people=people,dscs=dscs,types=db.execute(select(RegistrationType).where(RegistrationType.is_active.is_(True)).order_by(RegistrationType.sort_order)).scalars().all(),rules=rules,consultants=consultants,can_manage=_can_manage(db,user)))
@router.post("/clients/{client_id}/people")
async def add_person(request:Request,client_id:int,person_type:str=Form(...),full_name:str=Form(...),designation:str=Form(""),pan:str=Form(""),din:str=Form(""),date_of_birth:str=Form(""),appointment_date:str=Form(""),mobile:str=Form(""),email:str=Form(""),authorised_signatory:bool=Form(False),notes:str=Form(""),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");c=db.get(Client,client_id)
db.add(ClientRelatedPerson(tenant_id=c.tenant_id,client_id=c.id,person_type=person_type,full_name=full_name.strip(),designation=designation or None,pan=pan.strip().upper() or None,din=din or None,date_of_birth=_d(date_of_birth),appointment_date=_d(appointment_date),mobile=mobile or None,email=email or None,authorised_signatory=authorised_signatory,notes=notes or None,created_by_user_id=user.id));db.commit();return RedirectResponse(f'/registrations/clients/{client_id}',303)
@router.post("/clients/{client_id}/records")
async def add_registration(request:Request,client_id:int,registration_type_id:int=Form(...),related_person_id:str=Form(""),registration_number:str=Form(...),jurisdiction:str=Form(""),state_code:str=Form(""),issue_date:str=Form(""),effective_from:str=Form(""),valid_until:str=Form(""),next_action_date:str=Form(""),lifecycle_rule_id:str=Form(""),responsible_party:str=Form("firm"),assigned_user_id:str=Form(""),assigned_consultant_id:str=Form(""),auto_create_task:bool=Form(False),notes:str=Form(""),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");c=db.get(Client,client_id)
db.add(ClientRegistration(tenant_id=c.tenant_id,branch_id=c.branch_id,client_id=c.id,registration_type_id=registration_type_id,related_person_id=int(related_person_id) if related_person_id else None,registration_number=registration_number.strip().upper(),jurisdiction=jurisdiction or None,state_code=state_code or None,issue_date=_d(issue_date),effective_from=_d(effective_from),valid_until=_d(valid_until),next_action_date=_d(next_action_date),lifecycle_rule_id=int(lifecycle_rule_id) if lifecycle_rule_id else None,responsible_party=responsible_party,assigned_user_id=int(assigned_user_id) if assigned_user_id else c.partner_id,assigned_consultant_id=int(assigned_consultant_id) if assigned_consultant_id else None,auto_create_task=auto_create_task,notes=notes or None,created_by_user_id=user.id));db.commit();run_registration_lifecycle_once(db);return RedirectResponse(f'/registrations/clients/{client_id}',303)
@router.post("/clients/{client_id}/dsc")
async def add_dsc(request:Request,client_id:int,related_person_id:int=Form(...),certificate_serial_number:str=Form(""),provider:str=Form(""),token_make:str=Form(""),class_type:str=Form(""),issued_on:str=Form(""),expires_on:str=Form(...),custody_location:str=Form(""),renewal_responsibility:str=Form("firm"),assigned_consultant_id:str=Form(""),notes:str=Form(""),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");c=db.get(Client,client_id)
expiry=_d(expires_on)
dsc=ClientDigitalSignature(tenant_id=c.tenant_id,client_id=c.id,related_person_id=related_person_id,certificate_serial_number=certificate_serial_number or None,provider=provider or None,token_make=token_make or None,class_type=class_type or None,issued_on=_d(issued_on),expires_on=expiry,custody_location=custody_location or None,renewal_responsibility=renewal_responsibility,assigned_consultant_id=int(assigned_consultant_id) if assigned_consultant_id else None,notes=notes or None,created_by_user_id=user.id)
db.add(dsc)
dsc_type=db.execute(select(RegistrationType).where(RegistrationType.code=="DSC")).scalar_one()
serial=(certificate_serial_number or f"DSC-{client_id}-{related_person_id}-{expiry.isoformat()}").strip().upper()
existing=db.execute(select(ClientRegistration).where(ClientRegistration.tenant_id==c.tenant_id,ClientRegistration.registration_type_id==dsc_type.id,ClientRegistration.registration_number==serial)).scalar_one_or_none()
if not existing:
db.add(ClientRegistration(tenant_id=c.tenant_id,branch_id=c.branch_id,client_id=c.id,registration_type_id=dsc_type.id,related_person_id=related_person_id,registration_number=serial,issue_date=_d(issued_on),effective_from=_d(issued_on),valid_until=expiry,responsible_party=renewal_responsibility,assigned_user_id=c.partner_id,assigned_consultant_id=int(assigned_consultant_id) if assigned_consultant_id else None,auto_create_task=True,notes=notes or None,created_by_user_id=user.id))
db.commit();run_registration_lifecycle_once(db);return RedirectResponse(f'/registrations/clients/{client_id}',303)
@router.get("/tasks/{task_id}",response_class=HTMLResponse)
def task_detail(request:Request,task_id:int):
with CommonSessionLocal() as db:
user=_user(request,db)
if not user:return RedirectResponse('/login',303)
task=db.get(RegistrationRenewalTask,task_id)
if not task or task.tenant_id!=_tenant(request,user):return RedirectResponse('/registrations',303)
reg=db.get(ClientRegistration,task.registration_id);client=db.get(Client,task.client_id);rt=db.get(RegistrationType,reg.registration_type_id)
docs=db.execute(select(RegistrationRenewalTaskDocument).where(RegistrationRenewalTaskDocument.renewal_task_id==task.id)).scalars().all()
actions=db.execute(select(ClientRegistrationAction).where(ClientRegistrationAction.registration_id==reg.id).order_by(ClientRegistrationAction.created_at_utc.desc())).scalars().all()
return templates.TemplateResponse("modules/registrations/templates/registrations/task.html",_ctx(request,user,db,task=task,registration=reg,client=client,registration_type=rt,documents=docs,actions=actions,can_manage=_can_manage(db,user)))
@router.post("/tasks/{task_id}/status")
async def update_task(request:Request,task_id:int,status:str=Form(...),submitted_on:str=Form(""),acknowledgement_number:str=Form(""),notes:str=Form(""),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");task=db.get(RegistrationRenewalTask,task_id);task.status=status;task.submitted_on=_d(submitted_on);task.acknowledgement_number=acknowledgement_number or None;task.notes=notes or None;db.commit();return RedirectResponse(f'/registrations/tasks/{task_id}',303)
@router.post("/tasks/{task_id}/complete")
async def complete(request:Request,task_id:int,new_valid_until:str=Form(""),acknowledgement_number:str=Form(""),notes:str=Form(""),csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");task=db.get(RegistrationRenewalTask,task_id);complete_renewal(db,task,new_valid_until=_d(new_valid_until),acknowledgement_number=acknowledgement_number or None,actor_user_id=user.id,notes=notes or None);return RedirectResponse(f'/registrations/tasks/{task_id}',303)
@router.post("/run-engine")
async def run_engine(request:Request,csrf_token:str=Form(...)):
with CommonSessionLocal() as db:
user=_user(request,db);validate_csrf(request,csrf_token);require_permission(db,user,"clients.edit");run_registration_lifecycle_once(db);return RedirectResponse('/registrations',303)
Reference in New Issue View Git Blame Copy Permalink