diff --git a/alembic/versions/20260807_service_catalogue_registration_type_master.py b/alembic/versions/20260807_service_catalogue_registration_type_master.py new file mode 100644 index 0000000..e8e867a --- /dev/null +++ b/alembic/versions/20260807_service_catalogue_registration_type_master.py @@ -0,0 +1,45 @@ +"""Align Service Catalogue registration requirements with Registration Type master. + +The Service Catalogue previously used a frontend-only legacy code "GST" while the +Registration Type master uses "GSTIN". Scope mapping compares these codes exactly. +This migration updates only the legacy Service Catalogue value and leaves service +IDs, service codes, firm selections, subscriptions, engagements, tasks and all +other catalogue data unchanged. +""" + +from alembic import op +import sqlalchemy as sa + + +revision = "20260807_service_reg_master" +down_revision = "20260806_related_person_scope" +branch_labels = None +depends_on = None + + +def upgrade(): + bind = op.get_bind() + inspector = sa.inspect(bind) + + if "service_catalogues" not in inspector.get_table_names(): + return + + columns = {column["name"] for column in inspector.get_columns("service_catalogues")} + if "required_registration_type" not in columns: + return + + bind.execute( + sa.text( + """ + UPDATE service_catalogues + SET required_registration_type = 'GSTIN' + WHERE UPPER(TRIM(required_registration_type)) = 'GST' + """ + ) + ) + + +def downgrade(): + # Deliberately do not restore GST. GSTIN is the canonical Registration Type + # master code and reverting it would reintroduce the scope-map mismatch. + pass diff --git a/app/modules/services/templates/services/catalogue_form.html b/app/modules/services/templates/services/catalogue_form.html index 1015ea9..a13312f 100644 --- a/app/modules/services/templates/services/catalogue_form.html +++ b/app/modules/services/templates/services/catalogue_form.html @@ -19,11 +19,18 @@
-

Used only for Registration-level services.

+

Used only for Registration-level services. Values come from the active Registration Type Master.

+ {% if request.query_params.get('error') == 'registration_type' %} +

Select an active Registration Type Master value.

+ {% endif %}
diff --git a/app/modules/services/ui.py b/app/modules/services/ui.py index f76b94a..a3267b4 100644 --- a/app/modules/services/ui.py +++ b/app/modules/services/ui.py @@ -14,6 +14,7 @@ from app.modules.core.audit.service import model_snapshot, pair_before_after, wr 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.core.tenancy.models import Branch +from app.modules.registrations.models import RegistrationType from app.modules.services.models import FirmServiceSelection, FirmServiceTaskTemplate, ServiceCatalogue, ServiceCategory, ServiceDefaultTaskTemplate, ServiceDueDateRule, FirmTaskDocumentRequirement, FirmTaskDocumentTemplate from app.modules.services.bulk_imports import ( build_template as build_bulk_import_template, @@ -87,6 +88,50 @@ def _render(request: Request, template: str, db, user, **ctx): return templates.TemplateResponse(template, _base_ctx(request, user, db, **ctx)) +def _active_registration_types(db): + """Return the active Registration Type master rows used by Service Catalogue forms.""" + return db.execute( + select(RegistrationType) + .where(RegistrationType.is_active.is_(True)) + .order_by(RegistrationType.sort_order, RegistrationType.name, RegistrationType.code) + ).scalars().all() + + +def _registration_type_form_context(db): + registration_types = _active_registration_types(db) + return { + "registration_types": registration_types, + "registration_type_codes": [row.code.upper() for row in registration_types], + } + + +def _validated_registration_type_code(db, value: str, *, existing_value: str | None = None) -> str | None: + """Validate a submitted catalogue registration code against Registration Type master. + + Existing non-master legacy values are allowed only when left unchanged so editing an + unrelated field never destroys historical catalogue data. New/changed values must be + active Registration Type master codes. + """ + code = (value or "").strip().upper() + if not code: + return None + + exists = db.execute( + select(RegistrationType.id).where( + RegistrationType.code == code, + RegistrationType.is_active.is_(True), + ) + ).scalar_one_or_none() + if exists is not None: + return code + + current = (existing_value or "").strip().upper() + if current and code == current: + return code + + raise ValueError("registration_type") + + def _redirect_denied(): from app.core.http_responses import ui_access_denied return ui_access_denied() @@ -476,7 +521,7 @@ def catalogue_create_page(request: Request): if not _is_system_admin(db, user): return _redirect_denied() require_permission(db, user, 'services.create') - return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Create Service Catalogue', mode='create', catalogue=None, categories=list_categories(db)) + return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Create Service Catalogue', mode='create', catalogue=None, categories=list_categories(db), **_registration_type_form_context(db)) except Exception: return _redirect_denied() finally: @@ -495,6 +540,10 @@ def catalogue_create_submit(request: Request, service_code: str = Form(...), ser return _redirect_denied() require_permission(db, user, 'services.create') selected_category = get_category(db, int(category_id)) if str(category_id).strip() else None + try: + registration_type_code = _validated_registration_type_code(db, required_registration_type) + except ValueError: + return RedirectResponse(url='/services/catalogue/new?error=registration_type', status_code=303) row = ServiceCatalogue( service_code=normalize_code(service_code), service_name=service_name.strip(), @@ -503,7 +552,7 @@ def catalogue_create_submit(request: Request, service_code: str = Form(...), ser recurrence_type=recurrence_type.strip() or None, engagement_type=normalize_engagement_type(engagement_type), service_scope_type=(service_scope_type or 'client').strip().lower(), - required_registration_type=(required_registration_type or '').strip().upper() or None, + required_registration_type=registration_type_code, sort_order=sort_order, description=description.strip() or None, applicable_individual=applicable_individual is not None, @@ -806,7 +855,7 @@ def catalogue_edit_page(request: Request, catalogue_id: int): row = get_catalogue(db, catalogue_id) if not row or not row.is_active: return RedirectResponse(url='/services/catalogue', status_code=303) - return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Edit Service Catalogue', mode='edit', catalogue=row, categories=list_categories(db)) + return _render(request, 'modules/services/templates/services/catalogue_form.html', db, user, title='Edit Service Catalogue', mode='edit', catalogue=row, categories=list_categories(db), **_registration_type_form_context(db)) except Exception: return _redirect_denied() finally: @@ -828,13 +877,22 @@ def catalogue_edit_submit(request: Request, catalogue_id: int, service_name: str if not row or not row.is_active: return RedirectResponse(url='/services/catalogue', status_code=303) selected_category = get_category(db, int(category_id)) if str(category_id).strip() else None + try: + registration_type_code = _validated_registration_type_code( + db, required_registration_type, existing_value=row.required_registration_type + ) + except ValueError: + return RedirectResponse( + url=f'/services/catalogue/{catalogue_id}/edit?error=registration_type', + status_code=303, + ) row.service_name = service_name.strip() row.category_id = selected_category.id if selected_category else None row.category = selected_category.name if selected_category else None row.recurrence_type = recurrence_type.strip() or None row.engagement_type = normalize_engagement_type(engagement_type) row.service_scope_type = (service_scope_type or 'client').strip().lower() - row.required_registration_type = (required_registration_type or '').strip().upper() or None + row.required_registration_type = registration_type_code row.sort_order = sort_order row.applicable_individual = applicable_individual is not None row.applicable_proprietorship = applicable_proprietorship is not None