Seamless User Onboarding with Hyrex

1from hyrex import HyrexRegistry
2import smtplib
3from email.mime.text import MIMEText
4from email.mime.multipart import MIMEMultipart
5import jinja2
6import stripe
7
8hy = HyrexRegistry()
9
10@hy.task
11def send_welcome_email(user_id: str, user_email: str, user_name: str):
12    template = jinja2.Template("""
13    Hi {{ user_name }},
14
15    Welcome to our platform! Here's what you can do next:
16    - Complete your profile setup
17    - Explore our dashboard
18    - Connect with your team
19
20    Best regards,
21    The Team
22    """)
23
24    msg = MIMEMultipart()
25    msg['From'] = os.environ.get('SMTP_FROM_EMAIL')
26    msg['To'] = user_email
27    msg['Subject'] = "Welcome aboard!"
28
29    body = template.render(user_name=user_name)
30    msg.attach(MIMEText(body, 'plain'))
31
32    server = smtplib.SMTP('smtp.gmail.com', 587)
33    server.starttls()
34    server.login(os.environ.get('SMTP_USERNAME'), os.environ.get('SMTP_PASSWORD'))
35    server.send_message(msg)
36    server.quit()
37
38@hy.task
39def setup_user_trial(user_id: str, plan_type: str = "basic"):
40    # Create Stripe customer and trial subscription
41    customer = stripe.Customer.create(
42        metadata={"user_id": user_id}
43    )
44
45    stripe.Subscription.create(
46        customer=customer.id,
47        items=[{"price": f"price_{plan_type}_monthly"}],
48        trial_period_days=14,
49        metadata={"user_id": user_id}
50    )
51
52    return {"customer_id": customer.id, "trial_days": 14}
53
54@hy.task
55def create_sample_data(user_id: str):
56    # Generate sample data for new users
57    sample_projects = [
58        {"name": "My First Project", "description": "Get started here"},
59        {"name": "Demo Workflow", "description": "See how it works"}
60    ]
61
62    for project in sample_projects:
63        create_project.send(user_id, project)
64
65@hy.task
66def onboard_new_user(user_id: str, user_email: str, user_name: str, plan_type: str = "basic"):
67    # Orchestrate the full onboarding process
68    send_welcome_email.send(user_id, user_email, user_name)
69    trial_info = setup_user_trial.send(user_id, plan_type)
70    create_sample_data.send(user_id)
71
72    return {"status": "onboarding_started", "trial_info": trial_info}

1from fastapi import FastAPI
2from pydantic import BaseModel
3from .tasks import onboard_new_user, send_welcome_email, setup_user_trial
4
5app = FastAPI()
6
7class UserOnboardingRequest(BaseModel):
8    user_id: str
9    user_email: str
10    user_name: str
11    plan_type: str = "basic"
12
13class WelcomeEmailRequest(BaseModel):
14    user_id: str
15    user_email: str
16    user_name: str
17
18@app.post("/onboarding/start")
19async def start_user_onboarding(request: UserOnboardingRequest):
20    # Trigger complete onboarding workflow
21    task = onboard_new_user.send(
22        request.user_id,
23        request.user_email,
24        request.user_name,
25        request.plan_type
26    )
27    return {"message": "Onboarding started", "task_id": task.task_id}
28
29@app.post("/onboarding/welcome-email")
30async def send_welcome_email_only(request: WelcomeEmailRequest):
31    # Send just the welcome email
32    task = send_welcome_email.send(
33        request.user_id,
34        request.user_email,
35        request.user_name
36    )
37    return {"message": "Welcome email queued", "task_id": task.task_id}
38
39@app.post("/onboarding/setup-trial/{user_id}")
40async def setup_trial_only(user_id: str, plan_type: str = "basic"):
41    # Setup trial subscription only
42    task = setup_user_trial.send(user_id, plan_type)
43    return {"message": "Trial setup started", "task_id": task.task_id}