forked from andika/membership-be
1. Database Migration (backend/alembic/versions/014_add_custom_registration_data.py)- Adds custom_registration_data JSON column to users table for storing dynamic field responses2. User Model (backend/models.py)- Added custom_registration_data = Column(JSON, default=dict, nullable=False) to User model3. New API Endpoints (backend/server.py)- GET /api/registration/schema - Public endpoint returning form schema- GET /api/admin/registration/schema - Admin view with metadata- PUT /api/admin/registration/schema - Update schema- POST /api/admin/registration/schema/validate - Validate schema structure- POST /api/admin/registration/schema/reset - Reset to default- GET /api/admin/registration/field-types - Get available field types4. Validation Functions- validate_dynamic_registration() - Validates form data against schema- split_registration_data() - Splits data between User columns and custom_registration_data- evaluate_conditional_rules() - Evaluates show/hide rules5. Permissions (backend/seed_permissions_rbac.py)- Added registration.view and registration.manage permissions
This commit is contained in:
Koncept Kit
807
server.py
807
server.py
@@ -134,16 +134,23 @@ def set_user_role(user: User, role_enum: UserRole, db: Session):
|
||||
# Pydantic Models
|
||||
# ============================================================
|
||||
class RegisterRequest(BaseModel):
|
||||
# Step 1: Personal & Partner Information
|
||||
"""Dynamic registration request - validates against registration schema"""
|
||||
|
||||
# Fixed required fields (always present)
|
||||
first_name: str
|
||||
last_name: str
|
||||
phone: str
|
||||
address: str
|
||||
city: str
|
||||
state: str
|
||||
zipcode: str
|
||||
date_of_birth: datetime
|
||||
lead_sources: List[str]
|
||||
email: EmailStr
|
||||
password: str = Field(min_length=6)
|
||||
accepts_tos: bool = False
|
||||
|
||||
# Step 1: Personal & Partner Information (optional for dynamic schema)
|
||||
phone: Optional[str] = None
|
||||
address: Optional[str] = None
|
||||
city: Optional[str] = None
|
||||
state: Optional[str] = None
|
||||
zipcode: Optional[str] = None
|
||||
date_of_birth: Optional[datetime] = None
|
||||
lead_sources: Optional[List[str]] = None
|
||||
partner_first_name: Optional[str] = None
|
||||
partner_last_name: Optional[str] = None
|
||||
partner_is_member: Optional[bool] = False
|
||||
@@ -151,16 +158,16 @@ class RegisterRequest(BaseModel):
|
||||
|
||||
# Step 2: Newsletter, Volunteer & Scholarship
|
||||
referred_by_member_name: Optional[str] = None
|
||||
newsletter_publish_name: bool
|
||||
newsletter_publish_photo: bool
|
||||
newsletter_publish_birthday: bool
|
||||
newsletter_publish_none: bool
|
||||
volunteer_interests: List[str] = []
|
||||
scholarship_requested: bool = False
|
||||
newsletter_publish_name: Optional[bool] = False
|
||||
newsletter_publish_photo: Optional[bool] = False
|
||||
newsletter_publish_birthday: Optional[bool] = False
|
||||
newsletter_publish_none: Optional[bool] = False
|
||||
volunteer_interests: Optional[List[str]] = []
|
||||
scholarship_requested: Optional[bool] = False
|
||||
scholarship_reason: Optional[str] = None
|
||||
|
||||
# Step 3: Directory Settings
|
||||
show_in_directory: bool = False
|
||||
show_in_directory: Optional[bool] = False
|
||||
directory_email: Optional[str] = None
|
||||
directory_bio: Optional[str] = None
|
||||
directory_address: Optional[str] = None
|
||||
@@ -168,10 +175,9 @@ class RegisterRequest(BaseModel):
|
||||
directory_dob: Optional[datetime] = None
|
||||
directory_partner_name: Optional[str] = None
|
||||
|
||||
# Step 4: Account Credentials
|
||||
email: EmailStr
|
||||
password: str = Field(min_length=6)
|
||||
accepts_tos: bool = False
|
||||
# Allow extra fields for custom registration data
|
||||
class Config:
|
||||
extra = 'allow'
|
||||
|
||||
@validator('accepts_tos')
|
||||
def tos_must_be_accepted(cls, v):
|
||||
@@ -179,25 +185,6 @@ class RegisterRequest(BaseModel):
|
||||
raise ValueError('You must accept the Terms of Service to register')
|
||||
return v
|
||||
|
||||
@validator('newsletter_publish_none')
|
||||
def validate_newsletter_preferences(cls, v, values):
|
||||
"""At least one newsletter preference must be selected"""
|
||||
name = values.get('newsletter_publish_name', False)
|
||||
photo = values.get('newsletter_publish_photo', False)
|
||||
birthday = values.get('newsletter_publish_birthday', False)
|
||||
|
||||
if not (name or photo or birthday or v):
|
||||
raise ValueError('At least one newsletter publication preference must be selected')
|
||||
return v
|
||||
|
||||
@validator('scholarship_reason')
|
||||
def validate_scholarship_reason(cls, v, values):
|
||||
"""If scholarship requested, reason must be provided"""
|
||||
requested = values.get('scholarship_requested', False)
|
||||
if requested and not v:
|
||||
raise ValueError('Scholarship reason is required when requesting scholarship')
|
||||
return v
|
||||
|
||||
class LoginRequest(BaseModel):
|
||||
email: EmailStr
|
||||
password: str
|
||||
@@ -560,11 +547,28 @@ async def register(request: RegisterRequest, db: Session = Depends(get_db)):
|
||||
existing_user = db.query(User).filter(User.email == request.email).first()
|
||||
if existing_user:
|
||||
raise HTTPException(status_code=400, detail="Email already registered")
|
||||
|
||||
|
||||
# Get registration schema for dynamic validation
|
||||
schema = get_registration_schema(db)
|
||||
|
||||
# Convert request to dict for dynamic validation
|
||||
request_data = request.dict(exclude_unset=False)
|
||||
|
||||
# Perform dynamic schema validation
|
||||
is_valid, validation_errors = validate_dynamic_registration(request_data, schema)
|
||||
if not is_valid:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={"message": "Validation failed", "errors": validation_errors}
|
||||
)
|
||||
|
||||
# Split data into User model fields and custom fields
|
||||
user_data, custom_data = split_registration_data(request_data, schema)
|
||||
|
||||
# Generate verification token
|
||||
verification_token = secrets.token_urlsafe(32)
|
||||
|
||||
# Create user
|
||||
|
||||
# Create user with known fields
|
||||
user = User(
|
||||
# Account credentials (Step 4)
|
||||
email=request.email,
|
||||
@@ -573,65 +577,68 @@ async def register(request: RegisterRequest, db: Session = Depends(get_db)):
|
||||
# Personal information (Step 1)
|
||||
first_name=request.first_name,
|
||||
last_name=request.last_name,
|
||||
phone=request.phone,
|
||||
address=request.address,
|
||||
city=request.city,
|
||||
state=request.state,
|
||||
zipcode=request.zipcode,
|
||||
date_of_birth=request.date_of_birth,
|
||||
lead_sources=request.lead_sources,
|
||||
phone=user_data.get('phone') or request.phone,
|
||||
address=user_data.get('address') or request.address,
|
||||
city=user_data.get('city') or request.city,
|
||||
state=user_data.get('state') or request.state,
|
||||
zipcode=user_data.get('zipcode') or request.zipcode,
|
||||
date_of_birth=user_data.get('date_of_birth') or request.date_of_birth,
|
||||
lead_sources=user_data.get('lead_sources') or request.lead_sources or [],
|
||||
|
||||
# Partner information (Step 1)
|
||||
partner_first_name=request.partner_first_name,
|
||||
partner_last_name=request.partner_last_name,
|
||||
partner_is_member=request.partner_is_member,
|
||||
partner_plan_to_become_member=request.partner_plan_to_become_member,
|
||||
partner_first_name=user_data.get('partner_first_name') or request.partner_first_name,
|
||||
partner_last_name=user_data.get('partner_last_name') or request.partner_last_name,
|
||||
partner_is_member=user_data.get('partner_is_member', request.partner_is_member) or False,
|
||||
partner_plan_to_become_member=user_data.get('partner_plan_to_become_member', request.partner_plan_to_become_member) or False,
|
||||
|
||||
# Referral (Step 2)
|
||||
referred_by_member_name=request.referred_by_member_name,
|
||||
referred_by_member_name=user_data.get('referred_by_member_name') or request.referred_by_member_name,
|
||||
|
||||
# Newsletter publication preferences (Step 2)
|
||||
newsletter_publish_name=request.newsletter_publish_name,
|
||||
newsletter_publish_photo=request.newsletter_publish_photo,
|
||||
newsletter_publish_birthday=request.newsletter_publish_birthday,
|
||||
newsletter_publish_none=request.newsletter_publish_none,
|
||||
newsletter_publish_name=user_data.get('newsletter_publish_name', request.newsletter_publish_name) or False,
|
||||
newsletter_publish_photo=user_data.get('newsletter_publish_photo', request.newsletter_publish_photo) or False,
|
||||
newsletter_publish_birthday=user_data.get('newsletter_publish_birthday', request.newsletter_publish_birthday) or False,
|
||||
newsletter_publish_none=user_data.get('newsletter_publish_none', request.newsletter_publish_none) or False,
|
||||
|
||||
# Volunteer interests (Step 2)
|
||||
volunteer_interests=request.volunteer_interests,
|
||||
volunteer_interests=user_data.get('volunteer_interests') or request.volunteer_interests or [],
|
||||
|
||||
# Scholarship (Step 2)
|
||||
scholarship_requested=request.scholarship_requested,
|
||||
scholarship_reason=request.scholarship_reason,
|
||||
scholarship_requested=user_data.get('scholarship_requested', request.scholarship_requested) or False,
|
||||
scholarship_reason=user_data.get('scholarship_reason') or request.scholarship_reason,
|
||||
|
||||
# Directory settings (Step 3)
|
||||
show_in_directory=request.show_in_directory,
|
||||
directory_email=request.directory_email,
|
||||
directory_bio=request.directory_bio,
|
||||
directory_address=request.directory_address,
|
||||
directory_phone=request.directory_phone,
|
||||
directory_dob=request.directory_dob,
|
||||
directory_partner_name=request.directory_partner_name,
|
||||
show_in_directory=user_data.get('show_in_directory', request.show_in_directory) or False,
|
||||
directory_email=user_data.get('directory_email') or request.directory_email,
|
||||
directory_bio=user_data.get('directory_bio') or request.directory_bio,
|
||||
directory_address=user_data.get('directory_address') or request.directory_address,
|
||||
directory_phone=user_data.get('directory_phone') or request.directory_phone,
|
||||
directory_dob=user_data.get('directory_dob') or request.directory_dob,
|
||||
directory_partner_name=user_data.get('directory_partner_name') or request.directory_partner_name,
|
||||
|
||||
# Terms of Service acceptance (Step 4)
|
||||
accepts_tos=request.accepts_tos,
|
||||
tos_accepted_at=datetime.now(timezone.utc) if request.accepts_tos else None,
|
||||
|
||||
# Custom registration data for dynamic fields
|
||||
custom_registration_data=custom_data if custom_data else {},
|
||||
|
||||
# Status fields
|
||||
status=UserStatus.pending_email,
|
||||
role=UserRole.guest,
|
||||
email_verified=False,
|
||||
email_verification_token=verification_token
|
||||
)
|
||||
|
||||
|
||||
db.add(user)
|
||||
db.commit()
|
||||
db.refresh(user)
|
||||
|
||||
|
||||
# Send verification email
|
||||
await send_verification_email(user.email, verification_token)
|
||||
|
||||
|
||||
logger.info(f"User registered: {user.email}")
|
||||
|
||||
|
||||
return {"message": "Registration successful. Please check your email to verify your account."}
|
||||
|
||||
@api_router.get("/auth/verify-email")
|
||||
@@ -2062,6 +2069,664 @@ async def get_config_limits():
|
||||
"max_storage_bytes": int(os.getenv('MAX_STORAGE_BYTES', 1073741824))
|
||||
}
|
||||
|
||||
# ============================================================================
|
||||
# Registration Form Schema Routes
|
||||
# ============================================================================
|
||||
|
||||
# Default registration schema matching current 4-step form
|
||||
DEFAULT_REGISTRATION_SCHEMA = {
|
||||
"version": "1.0",
|
||||
"steps": [
|
||||
{
|
||||
"id": "step_personal",
|
||||
"title": "Personal Information",
|
||||
"description": "Please provide your personal details and tell us how you heard about us.",
|
||||
"order": 1,
|
||||
"sections": [
|
||||
{
|
||||
"id": "section_personal_info",
|
||||
"title": "Personal Information",
|
||||
"order": 1,
|
||||
"fields": [
|
||||
{"id": "first_name", "type": "text", "label": "First Name", "required": True, "is_fixed": True, "mapping": "first_name", "validation": {"minLength": 1, "maxLength": 100}, "width": "half", "order": 1},
|
||||
{"id": "last_name", "type": "text", "label": "Last Name", "required": True, "is_fixed": True, "mapping": "last_name", "validation": {"minLength": 1, "maxLength": 100}, "width": "half", "order": 2},
|
||||
{"id": "phone", "type": "phone", "label": "Phone", "required": True, "is_fixed": False, "mapping": "phone", "width": "half", "order": 3},
|
||||
{"id": "date_of_birth", "type": "date", "label": "Date of Birth", "required": True, "is_fixed": False, "mapping": "date_of_birth", "width": "half", "order": 4},
|
||||
{"id": "address", "type": "text", "label": "Address", "required": True, "is_fixed": False, "mapping": "address", "width": "full", "order": 5},
|
||||
{"id": "city", "type": "text", "label": "City", "required": True, "is_fixed": False, "mapping": "city", "width": "third", "order": 6},
|
||||
{"id": "state", "type": "text", "label": "State", "required": True, "is_fixed": False, "mapping": "state", "width": "third", "order": 7},
|
||||
{"id": "zipcode", "type": "text", "label": "Zipcode", "required": True, "is_fixed": False, "mapping": "zipcode", "width": "third", "order": 8}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "section_lead_sources",
|
||||
"title": "How Did You Hear About Us?",
|
||||
"order": 2,
|
||||
"fields": [
|
||||
{"id": "lead_sources", "type": "multiselect", "label": "How did you hear about us?", "required": True, "is_fixed": False, "mapping": "lead_sources", "width": "full", "order": 1, "options": [
|
||||
{"value": "Current member", "label": "Current member"},
|
||||
{"value": "Friend", "label": "Friend"},
|
||||
{"value": "OutSmart Magazine", "label": "OutSmart Magazine"},
|
||||
{"value": "Search engine (Google etc.)", "label": "Search engine (Google etc.)"},
|
||||
{"value": "I've known about LOAF for a long time", "label": "I've known about LOAF for a long time"},
|
||||
{"value": "Other", "label": "Other"}
|
||||
]}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "section_partner",
|
||||
"title": "Partner Information (Optional)",
|
||||
"order": 3,
|
||||
"fields": [
|
||||
{"id": "partner_first_name", "type": "text", "label": "Partner First Name", "required": False, "is_fixed": False, "mapping": "partner_first_name", "width": "half", "order": 1},
|
||||
{"id": "partner_last_name", "type": "text", "label": "Partner Last Name", "required": False, "is_fixed": False, "mapping": "partner_last_name", "width": "half", "order": 2},
|
||||
{"id": "partner_is_member", "type": "checkbox", "label": "Is your partner already a member?", "required": False, "is_fixed": False, "mapping": "partner_is_member", "width": "full", "order": 3},
|
||||
{"id": "partner_plan_to_become_member", "type": "checkbox", "label": "Does your partner plan to become a member?", "required": False, "is_fixed": False, "mapping": "partner_plan_to_become_member", "width": "full", "order": 4}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "step_newsletter",
|
||||
"title": "Newsletter & Volunteer",
|
||||
"description": "Tell us about your newsletter preferences and volunteer interests.",
|
||||
"order": 2,
|
||||
"sections": [
|
||||
{
|
||||
"id": "section_referral",
|
||||
"title": "Referral",
|
||||
"order": 1,
|
||||
"fields": [
|
||||
{"id": "referred_by_member_name", "type": "text", "label": "If referred by a current member, please provide their name", "required": False, "is_fixed": False, "mapping": "referred_by_member_name", "width": "full", "order": 1, "placeholder": "Enter member name or email"}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "section_newsletter_prefs",
|
||||
"title": "Newsletter Publication Preferences",
|
||||
"description": "Select what you would like published in our newsletter.",
|
||||
"order": 2,
|
||||
"fields": [
|
||||
{"id": "newsletter_publish_name", "type": "checkbox", "label": "Publish my name", "required": False, "is_fixed": False, "mapping": "newsletter_publish_name", "width": "full", "order": 1},
|
||||
{"id": "newsletter_publish_photo", "type": "checkbox", "label": "Publish my photo", "required": False, "is_fixed": False, "mapping": "newsletter_publish_photo", "width": "full", "order": 2},
|
||||
{"id": "newsletter_publish_birthday", "type": "checkbox", "label": "Publish my birthday", "required": False, "is_fixed": False, "mapping": "newsletter_publish_birthday", "width": "full", "order": 3},
|
||||
{"id": "newsletter_publish_none", "type": "checkbox", "label": "Don't publish anything about me", "required": False, "is_fixed": False, "mapping": "newsletter_publish_none", "width": "full", "order": 4}
|
||||
],
|
||||
"validation": {"atLeastOne": True, "message": "Please select at least one newsletter publication preference"}
|
||||
},
|
||||
{
|
||||
"id": "section_volunteer",
|
||||
"title": "Volunteer Interests",
|
||||
"order": 3,
|
||||
"fields": [
|
||||
{"id": "volunteer_interests", "type": "multiselect", "label": "Select areas where you would like to volunteer", "required": False, "is_fixed": False, "mapping": "volunteer_interests", "width": "full", "order": 1, "options": [
|
||||
{"value": "Events", "label": "Events"},
|
||||
{"value": "Hospitality", "label": "Hospitality"},
|
||||
{"value": "Newsletter", "label": "Newsletter"},
|
||||
{"value": "Board", "label": "Board"},
|
||||
{"value": "Community Outreach", "label": "Community Outreach"},
|
||||
{"value": "Other", "label": "Other"}
|
||||
]}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "section_scholarship",
|
||||
"title": "Scholarship Request",
|
||||
"order": 4,
|
||||
"fields": [
|
||||
{"id": "scholarship_requested", "type": "checkbox", "label": "I would like to request a scholarship", "required": False, "is_fixed": False, "mapping": "scholarship_requested", "width": "full", "order": 1},
|
||||
{"id": "scholarship_reason", "type": "textarea", "label": "Please explain why you are requesting a scholarship", "required": False, "is_fixed": False, "mapping": "scholarship_reason", "width": "full", "order": 2, "rows": 4}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "step_directory",
|
||||
"title": "Member Directory",
|
||||
"description": "Choose what information to display in the member directory.",
|
||||
"order": 3,
|
||||
"sections": [
|
||||
{
|
||||
"id": "section_directory_settings",
|
||||
"title": "Directory Settings",
|
||||
"order": 1,
|
||||
"fields": [
|
||||
{"id": "show_in_directory", "type": "checkbox", "label": "Show my profile in the member directory", "required": False, "is_fixed": False, "mapping": "show_in_directory", "width": "full", "order": 1},
|
||||
{"id": "directory_email", "type": "email", "label": "Directory Email (if different from account email)", "required": False, "is_fixed": False, "mapping": "directory_email", "width": "full", "order": 2},
|
||||
{"id": "directory_bio", "type": "textarea", "label": "Bio for directory", "required": False, "is_fixed": False, "mapping": "directory_bio", "width": "full", "order": 3, "rows": 4},
|
||||
{"id": "directory_address", "type": "text", "label": "Address to display in directory", "required": False, "is_fixed": False, "mapping": "directory_address", "width": "full", "order": 4},
|
||||
{"id": "directory_phone", "type": "phone", "label": "Phone to display in directory", "required": False, "is_fixed": False, "mapping": "directory_phone", "width": "half", "order": 5},
|
||||
{"id": "directory_dob", "type": "date", "label": "Birthday to display in directory", "required": False, "is_fixed": False, "mapping": "directory_dob", "width": "half", "order": 6},
|
||||
{"id": "directory_partner_name", "type": "text", "label": "Partner name to display in directory", "required": False, "is_fixed": False, "mapping": "directory_partner_name", "width": "full", "order": 7}
|
||||
]
|
||||
}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "step_account",
|
||||
"title": "Account Setup",
|
||||
"description": "Create your account credentials and accept the terms of service.",
|
||||
"order": 4,
|
||||
"sections": [
|
||||
{
|
||||
"id": "section_credentials",
|
||||
"title": "Account Credentials",
|
||||
"order": 1,
|
||||
"fields": [
|
||||
{"id": "email", "type": "email", "label": "Email Address", "required": True, "is_fixed": True, "mapping": "email", "width": "full", "order": 1},
|
||||
{"id": "password", "type": "password", "label": "Password", "required": True, "is_fixed": True, "mapping": "password", "validation": {"minLength": 6}, "width": "half", "order": 2},
|
||||
{"id": "confirmPassword", "type": "password", "label": "Confirm Password", "required": True, "is_fixed": True, "client_only": True, "width": "half", "order": 3, "validation": {"matchField": "password"}}
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "section_tos",
|
||||
"title": "Terms of Service",
|
||||
"order": 2,
|
||||
"fields": [
|
||||
{"id": "accepts_tos", "type": "checkbox", "label": "I accept the Terms of Service and Privacy Policy", "required": True, "is_fixed": True, "mapping": "accepts_tos", "width": "full", "order": 1}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"conditional_rules": [
|
||||
{
|
||||
"id": "rule_scholarship_reason",
|
||||
"trigger_field": "scholarship_requested",
|
||||
"trigger_operator": "equals",
|
||||
"trigger_value": True,
|
||||
"action": "show",
|
||||
"target_fields": ["scholarship_reason"]
|
||||
}
|
||||
],
|
||||
"fixed_fields": ["email", "password", "first_name", "last_name", "accepts_tos"]
|
||||
}
|
||||
|
||||
# Supported field types with their validation options
|
||||
FIELD_TYPES = {
|
||||
"text": {
|
||||
"name": "Text Input",
|
||||
"validation_options": ["required", "minLength", "maxLength", "pattern"],
|
||||
"properties": ["placeholder", "width"]
|
||||
},
|
||||
"email": {
|
||||
"name": "Email Input",
|
||||
"validation_options": ["required"],
|
||||
"properties": ["placeholder"]
|
||||
},
|
||||
"phone": {
|
||||
"name": "Phone Input",
|
||||
"validation_options": ["required"],
|
||||
"properties": ["placeholder"]
|
||||
},
|
||||
"date": {
|
||||
"name": "Date Input",
|
||||
"validation_options": ["required", "min_date", "max_date"],
|
||||
"properties": []
|
||||
},
|
||||
"dropdown": {
|
||||
"name": "Dropdown Select",
|
||||
"validation_options": ["required"],
|
||||
"properties": ["options", "placeholder"]
|
||||
},
|
||||
"checkbox": {
|
||||
"name": "Checkbox",
|
||||
"validation_options": ["required"],
|
||||
"properties": []
|
||||
},
|
||||
"radio": {
|
||||
"name": "Radio Group",
|
||||
"validation_options": ["required"],
|
||||
"properties": ["options"]
|
||||
},
|
||||
"multiselect": {
|
||||
"name": "Multi-Select",
|
||||
"validation_options": ["required", "min_selections", "max_selections"],
|
||||
"properties": ["options"]
|
||||
},
|
||||
"address_group": {
|
||||
"name": "Address Group",
|
||||
"validation_options": ["required"],
|
||||
"properties": []
|
||||
},
|
||||
"textarea": {
|
||||
"name": "Text Area",
|
||||
"validation_options": ["required", "minLength", "maxLength"],
|
||||
"properties": ["rows", "placeholder"]
|
||||
},
|
||||
"file_upload": {
|
||||
"name": "File Upload",
|
||||
"validation_options": ["required", "file_types", "max_size"],
|
||||
"properties": ["allowed_types", "max_size_mb"]
|
||||
},
|
||||
"password": {
|
||||
"name": "Password Input",
|
||||
"validation_options": ["required", "minLength"],
|
||||
"properties": []
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
class RegistrationSchemaRequest(BaseModel):
|
||||
"""Request model for updating registration schema"""
|
||||
schema_data: dict
|
||||
|
||||
|
||||
def get_registration_schema(db: Session) -> dict:
|
||||
"""Get the current registration form schema from database or return default"""
|
||||
setting = db.query(SystemSettings).filter(
|
||||
SystemSettings.setting_key == "registration.form_schema"
|
||||
).first()
|
||||
|
||||
if setting and setting.setting_value:
|
||||
import json
|
||||
try:
|
||||
return json.loads(setting.setting_value)
|
||||
except json.JSONDecodeError:
|
||||
logger.error("Failed to parse registration schema from database")
|
||||
return DEFAULT_REGISTRATION_SCHEMA.copy()
|
||||
|
||||
return DEFAULT_REGISTRATION_SCHEMA.copy()
|
||||
|
||||
|
||||
def save_registration_schema(db: Session, schema: dict, user_id: Optional[uuid.UUID] = None) -> None:
|
||||
"""Save registration schema to database"""
|
||||
import json
|
||||
|
||||
setting = db.query(SystemSettings).filter(
|
||||
SystemSettings.setting_key == "registration.form_schema"
|
||||
).first()
|
||||
|
||||
schema_json = json.dumps(schema)
|
||||
|
||||
if setting:
|
||||
setting.setting_value = schema_json
|
||||
setting.updated_by = user_id
|
||||
setting.updated_at = datetime.now(timezone.utc)
|
||||
else:
|
||||
from models import SettingType
|
||||
setting = SystemSettings(
|
||||
setting_key="registration.form_schema",
|
||||
setting_value=schema_json,
|
||||
setting_type=SettingType.json,
|
||||
description="Dynamic registration form schema defining steps, fields, and validation rules",
|
||||
updated_by=user_id,
|
||||
is_sensitive=False
|
||||
)
|
||||
db.add(setting)
|
||||
|
||||
db.commit()
|
||||
|
||||
|
||||
def validate_schema(schema: dict) -> tuple[bool, list[str]]:
|
||||
"""Validate registration schema structure"""
|
||||
errors = []
|
||||
|
||||
# Check version
|
||||
if "version" not in schema:
|
||||
errors.append("Schema must have a version field")
|
||||
|
||||
# Check steps
|
||||
if "steps" not in schema or not isinstance(schema.get("steps"), list):
|
||||
errors.append("Schema must have a steps array")
|
||||
return False, errors
|
||||
|
||||
if len(schema["steps"]) == 0:
|
||||
errors.append("Schema must have at least one step")
|
||||
|
||||
if len(schema["steps"]) > 10:
|
||||
errors.append("Schema cannot have more than 10 steps")
|
||||
|
||||
# Check fixed fields exist
|
||||
fixed_fields = schema.get("fixed_fields", ["email", "password", "first_name", "last_name", "accepts_tos"])
|
||||
all_field_ids = set()
|
||||
|
||||
for step in schema.get("steps", []):
|
||||
if "id" not in step:
|
||||
errors.append(f"Step missing id")
|
||||
continue
|
||||
|
||||
if "sections" not in step or not isinstance(step.get("sections"), list):
|
||||
errors.append(f"Step {step.get('id')} must have sections array")
|
||||
continue
|
||||
|
||||
for section in step.get("sections", []):
|
||||
if "fields" not in section or not isinstance(section.get("fields"), list):
|
||||
errors.append(f"Section {section.get('id')} must have fields array")
|
||||
continue
|
||||
|
||||
for field in section.get("fields", []):
|
||||
if "id" not in field:
|
||||
errors.append(f"Field missing id in section {section.get('id')}")
|
||||
continue
|
||||
|
||||
all_field_ids.add(field["id"])
|
||||
|
||||
if "type" not in field:
|
||||
errors.append(f"Field {field['id']} missing type")
|
||||
|
||||
if field.get("type") not in FIELD_TYPES:
|
||||
errors.append(f"Field {field['id']} has invalid type: {field.get('type')}")
|
||||
|
||||
# Verify fixed fields are present
|
||||
for fixed_field in fixed_fields:
|
||||
if fixed_field not in all_field_ids:
|
||||
errors.append(f"Fixed field '{fixed_field}' must be present in schema")
|
||||
|
||||
# Field limit check
|
||||
if len(all_field_ids) > 100:
|
||||
errors.append("Schema cannot have more than 100 fields")
|
||||
|
||||
return len(errors) == 0, errors
|
||||
|
||||
|
||||
def evaluate_conditional_rules(form_data: dict, rules: list) -> set:
|
||||
"""Evaluate conditional rules and return set of visible field IDs"""
|
||||
visible_fields = set()
|
||||
|
||||
# Start with all fields visible
|
||||
for rule in rules:
|
||||
target_fields = rule.get("target_fields", [])
|
||||
if rule.get("action") == "hide":
|
||||
visible_fields.update(target_fields)
|
||||
|
||||
# Apply rules
|
||||
for rule in rules:
|
||||
trigger_field = rule.get("trigger_field")
|
||||
trigger_value = rule.get("trigger_value")
|
||||
trigger_operator = rule.get("trigger_operator", "equals")
|
||||
action = rule.get("action", "show")
|
||||
target_fields = rule.get("target_fields", [])
|
||||
|
||||
field_value = form_data.get(trigger_field)
|
||||
|
||||
# Evaluate condition
|
||||
condition_met = False
|
||||
if trigger_operator == "equals":
|
||||
condition_met = field_value == trigger_value
|
||||
elif trigger_operator == "not_equals":
|
||||
condition_met = field_value != trigger_value
|
||||
elif trigger_operator == "contains":
|
||||
condition_met = trigger_value in (field_value or []) if isinstance(field_value, list) else trigger_value in str(field_value or "")
|
||||
elif trigger_operator == "not_empty":
|
||||
condition_met = bool(field_value)
|
||||
elif trigger_operator == "empty":
|
||||
condition_met = not bool(field_value)
|
||||
|
||||
# Apply action
|
||||
if condition_met:
|
||||
if action == "show":
|
||||
visible_fields.update(target_fields)
|
||||
elif action == "hide":
|
||||
visible_fields -= set(target_fields)
|
||||
|
||||
return visible_fields
|
||||
|
||||
|
||||
def validate_field_by_type(field: dict, value) -> list[str]:
|
||||
"""Validate a field value based on its type and validation rules"""
|
||||
errors = []
|
||||
field_type = field.get("type")
|
||||
validation = field.get("validation", {})
|
||||
label = field.get("label", field.get("id"))
|
||||
|
||||
if field_type == "text" or field_type == "textarea":
|
||||
if not isinstance(value, str):
|
||||
errors.append(f"{label} must be text")
|
||||
return errors
|
||||
if "minLength" in validation and len(value) < validation["minLength"]:
|
||||
errors.append(f"{label} must be at least {validation['minLength']} characters")
|
||||
if "maxLength" in validation and len(value) > validation["maxLength"]:
|
||||
errors.append(f"{label} must be at most {validation['maxLength']} characters")
|
||||
if "pattern" in validation:
|
||||
import re
|
||||
if not re.match(validation["pattern"], value):
|
||||
errors.append(f"{label} format is invalid")
|
||||
|
||||
elif field_type == "email":
|
||||
import re
|
||||
email_pattern = r'^[^\s@]+@[^\s@]+\.[^\s@]+$'
|
||||
if not re.match(email_pattern, str(value)):
|
||||
errors.append(f"{label} must be a valid email address")
|
||||
|
||||
elif field_type == "phone":
|
||||
if not isinstance(value, str) or len(value) < 7:
|
||||
errors.append(f"{label} must be a valid phone number")
|
||||
|
||||
elif field_type == "date":
|
||||
# Date validation happens during parsing
|
||||
pass
|
||||
|
||||
elif field_type == "dropdown" or field_type == "radio":
|
||||
options = [opt.get("value") for opt in field.get("options", [])]
|
||||
if value not in options:
|
||||
errors.append(f"{label} must be one of the available options")
|
||||
|
||||
elif field_type == "multiselect":
|
||||
if not isinstance(value, list):
|
||||
errors.append(f"{label} must be a list of selections")
|
||||
else:
|
||||
options = [opt.get("value") for opt in field.get("options", [])]
|
||||
for v in value:
|
||||
if v not in options:
|
||||
errors.append(f"{label} contains invalid option: {v}")
|
||||
if "min_selections" in validation and len(value) < validation["min_selections"]:
|
||||
errors.append(f"{label} requires at least {validation['min_selections']} selections")
|
||||
if "max_selections" in validation and len(value) > validation["max_selections"]:
|
||||
errors.append(f"{label} allows at most {validation['max_selections']} selections")
|
||||
|
||||
elif field_type == "checkbox":
|
||||
if not isinstance(value, bool):
|
||||
errors.append(f"{label} must be true or false")
|
||||
|
||||
elif field_type == "password":
|
||||
if not isinstance(value, str):
|
||||
errors.append(f"{label} must be text")
|
||||
elif "minLength" in validation and len(value) < validation["minLength"]:
|
||||
errors.append(f"{label} must be at least {validation['minLength']} characters")
|
||||
|
||||
return errors
|
||||
|
||||
|
||||
def validate_dynamic_registration(data: dict, schema: dict) -> tuple[bool, list[str]]:
|
||||
"""Validate registration data against dynamic schema"""
|
||||
errors = []
|
||||
conditional_rules = schema.get("conditional_rules", [])
|
||||
|
||||
# Get all fields and their visibility based on conditional rules
|
||||
hidden_fields = set()
|
||||
for rule in conditional_rules:
|
||||
if rule.get("action") == "show":
|
||||
# Fields are hidden by default if they have a "show" rule
|
||||
hidden_fields.update(rule.get("target_fields", []))
|
||||
|
||||
# Evaluate which hidden fields should now be visible
|
||||
visible_conditional_fields = evaluate_conditional_rules(data, conditional_rules)
|
||||
hidden_fields -= visible_conditional_fields
|
||||
|
||||
for step in schema.get("steps", []):
|
||||
for section in step.get("sections", []):
|
||||
# Check section-level validation
|
||||
section_validation = section.get("validation", {})
|
||||
if section_validation.get("atLeastOne"):
|
||||
field_ids = [f["id"] for f in section.get("fields", [])]
|
||||
has_value = any(data.get(fid) for fid in field_ids)
|
||||
if not has_value:
|
||||
errors.append(section_validation.get("message", f"At least one field in {section.get('title', 'this section')} is required"))
|
||||
|
||||
for field in section.get("fields", []):
|
||||
field_id = field.get("id")
|
||||
|
||||
# Skip hidden fields
|
||||
if field_id in hidden_fields:
|
||||
continue
|
||||
|
||||
# Skip client-only fields (like confirmPassword)
|
||||
if field.get("client_only"):
|
||||
continue
|
||||
|
||||
value = data.get(field_id)
|
||||
|
||||
# Required check
|
||||
if field.get("required"):
|
||||
if value is None or value == "" or (isinstance(value, list) and len(value) == 0):
|
||||
errors.append(f"{field.get('label', field_id)} is required")
|
||||
continue
|
||||
|
||||
# Type-specific validation
|
||||
if value is not None and value != "":
|
||||
field_errors = validate_field_by_type(field, value)
|
||||
errors.extend(field_errors)
|
||||
|
||||
return len(errors) == 0, errors
|
||||
|
||||
|
||||
def split_registration_data(data: dict, schema: dict) -> tuple[dict, dict]:
|
||||
"""Split registration data into User model fields and custom fields"""
|
||||
user_data = {}
|
||||
custom_data = {}
|
||||
|
||||
# Get field mappings from schema
|
||||
field_mappings = {}
|
||||
for step in schema.get("steps", []):
|
||||
for section in step.get("sections", []):
|
||||
for field in section.get("fields", []):
|
||||
if field.get("mapping"):
|
||||
field_mappings[field["id"]] = field["mapping"]
|
||||
|
||||
# User model fields that have direct column mappings
|
||||
user_model_fields = {
|
||||
"email", "password", "first_name", "last_name", "phone", "address",
|
||||
"city", "state", "zipcode", "date_of_birth", "lead_sources",
|
||||
"partner_first_name", "partner_last_name", "partner_is_member",
|
||||
"partner_plan_to_become_member", "referred_by_member_name",
|
||||
"newsletter_publish_name", "newsletter_publish_photo",
|
||||
"newsletter_publish_birthday", "newsletter_publish_none",
|
||||
"volunteer_interests", "scholarship_requested", "scholarship_reason",
|
||||
"show_in_directory", "directory_email", "directory_bio",
|
||||
"directory_address", "directory_phone", "directory_dob",
|
||||
"directory_partner_name", "accepts_tos"
|
||||
}
|
||||
|
||||
for field_id, value in data.items():
|
||||
mapping = field_mappings.get(field_id, field_id)
|
||||
|
||||
# Skip client-only fields
|
||||
if field_id == "confirmPassword":
|
||||
continue
|
||||
|
||||
if mapping in user_model_fields:
|
||||
user_data[mapping] = value
|
||||
else:
|
||||
custom_data[field_id] = value
|
||||
|
||||
return user_data, custom_data
|
||||
|
||||
|
||||
# Public endpoint - returns schema for registration form
|
||||
@api_router.get("/registration/schema")
|
||||
async def get_public_registration_schema(db: Session = Depends(get_db)):
|
||||
"""Get registration form schema for public registration page"""
|
||||
schema = get_registration_schema(db)
|
||||
# Return a clean version without internal metadata
|
||||
return {
|
||||
"version": schema.get("version"),
|
||||
"steps": schema.get("steps", []),
|
||||
"conditional_rules": schema.get("conditional_rules", []),
|
||||
"fixed_fields": schema.get("fixed_fields", [])
|
||||
}
|
||||
|
||||
|
||||
# Admin endpoint - returns schema with metadata
|
||||
@api_router.get("/admin/registration/schema")
|
||||
async def get_admin_registration_schema(
|
||||
current_user: User = Depends(require_permission("registration.view")),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Get registration form schema with admin metadata"""
|
||||
schema = get_registration_schema(db)
|
||||
|
||||
# Get version info
|
||||
setting = db.query(SystemSettings).filter(
|
||||
SystemSettings.setting_key == "registration.form_schema"
|
||||
).first()
|
||||
|
||||
return {
|
||||
"schema": schema,
|
||||
"metadata": {
|
||||
"last_updated": setting.updated_at.isoformat() if setting else None,
|
||||
"updated_by": str(setting.updated_by) if setting and setting.updated_by else None,
|
||||
"is_default": setting is None
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
# Admin endpoint - update schema
|
||||
@api_router.put("/admin/registration/schema")
|
||||
async def update_registration_schema(
|
||||
request: RegistrationSchemaRequest,
|
||||
current_user: User = Depends(require_permission("registration.manage")),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Update registration form schema"""
|
||||
schema = request.schema_data
|
||||
|
||||
# Validate schema structure
|
||||
is_valid, errors = validate_schema(schema)
|
||||
if not is_valid:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail={"message": "Invalid schema", "errors": errors}
|
||||
)
|
||||
|
||||
# Save schema
|
||||
save_registration_schema(db, schema, current_user.id)
|
||||
|
||||
logger.info(f"Registration schema updated by user {current_user.email}")
|
||||
|
||||
return {"message": "Registration schema updated successfully"}
|
||||
|
||||
|
||||
# Admin endpoint - validate schema without saving
|
||||
@api_router.post("/admin/registration/schema/validate")
|
||||
async def validate_registration_schema_endpoint(
|
||||
request: RegistrationSchemaRequest,
|
||||
current_user: User = Depends(require_permission("registration.manage")),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Validate registration form schema without saving"""
|
||||
schema = request.schema_data
|
||||
is_valid, errors = validate_schema(schema)
|
||||
|
||||
return {
|
||||
"valid": is_valid,
|
||||
"errors": errors
|
||||
}
|
||||
|
||||
|
||||
# Admin endpoint - reset schema to default
|
||||
@api_router.post("/admin/registration/schema/reset")
|
||||
async def reset_registration_schema(
|
||||
current_user: User = Depends(require_permission("registration.manage")),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Reset registration form schema to default"""
|
||||
save_registration_schema(db, DEFAULT_REGISTRATION_SCHEMA.copy(), current_user.id)
|
||||
|
||||
logger.info(f"Registration schema reset to default by user {current_user.email}")
|
||||
|
||||
return {"message": "Registration schema reset to default"}
|
||||
|
||||
|
||||
# Admin endpoint - get available field types
|
||||
@api_router.get("/admin/registration/field-types")
|
||||
async def get_field_types(
|
||||
current_user: User = Depends(require_permission("registration.view")),
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""Get available field types for registration form builder"""
|
||||
return FIELD_TYPES
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# Admin Routes
|
||||
# ============================================================================
|
||||
|
||||
Reference in New Issue
Block a user