{
“project”: “Great World”,
“tech_stack”: “Python FastAPI”,
“is_initial_setup”: “False”,
“folder_structure”: null,
“user_story_output”: {
“story_id”: “APPLICANT-REG-001”,
“description”: “This feature implements a two-step user registration process. The first step captures the applicant’s email, mobile number, and password, validates the input, and sends a One-Time Password (OTP) to their mobile. The second step verifies the provided OTP to create and activate the user account, handling cases like incorrect OTPs, expired tokens, and duplicate user registrations.”,
“generated_files”: [
{
“path”: “src/api/v1/auth.py”,
“code_template”: “from fastapi import APIRouter, Depends, HTTPException, statusnfrom sqlalchemy.orm import Sessionnnfrom src.schemas.user_schemas import UserCreate, UserResponse, OTPVerifyRequestnfrom src.services import user_service, otp_servicenfrom src.db.session import get_dbnnrouter = APIRouter()[email protected](“/register”, status_code=status.HTTP_202_ACCEPTED, summary=”Initiate User Registration”)nasync def initiate_registration(user_in: UserCreate, db: Session = Depends(get_db)):n “””n Handles the first step of registration:n 1. Validates input data (email, mobile, password) via Pydantic model.n 2. Checks if a user with the given email or mobile already exists.n 3. Creates a temporary, inactive user record.n 4. Generates and sends an OTP to the user’s mobile number.n “””n db_user = user_service.get_user_by_email_or_mobile(db, email=user_in.email, mobile=user_in.mobile)n if db_user:n raise HTTPException(n status_code=status.HTTP_409_CONFLICT,n detail=”An account with this email or mobile number already exists.”,n )nn # Create an inactive user and generate OTPn # The service layer will handle the logic of not saving the password until verificationn otp_code = await user_service.create_inactive_user_and_send_otp(db, user_in=user_in)nn # In a real app, you would not return the OTP in the response.n # This is for demonstration/testing purposes.n return {“message”: “OTP sent to your mobile number for verification.”, “otp_for_testing”: otp_code}[email protected](“/verify-otp”, response_model=UserResponse, summary=”Verify OTP and Create Account”)nasync def verify_otp_and_create_user(otp_data: OTPVerifyRequest, db: Session = Depends(get_db)):n “””n Handles the second step of registration:n 1. Verifies the OTP against the user’s mobile number.n 2. If valid, activates the user account and marks it as verified.n 3. Returns the created user data and a login token.n “””n is_valid = await otp_service.verify_otp(db, mobile=otp_data.mobile, otp_code=otp_data.otp)n if not is_valid:n raise HTTPException(n status_code=status.HTTP_400_BAD_REQUEST,n detail=”Invalid or expired OTP. Please try again or request a new one.”,n )nn # If OTP is valid, the service activates the usern user = user_service.activate_user(db, mobile=otp_data.mobile)n if not user:n raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”User not found for activation.”)nn # Here you would typically generate a JWT token for loginn # token = security.create_access_token(data={“sub”: user.email})n # For now, we just return the user datan return [email protected](“/resend-otp”, status_code=status.HTTP_200_OK, summary=”Resend OTP”)nasync def resend_otp(mobile_data: dict, db: Session = Depends(get_db)):n “””n Allows the user to request a new OTP for a given mobile number.n – Rate limiting should be applied to this endpoint in a production environment.n “””n mobile = mobile_data.get(‘mobile’)n if not mobile:n raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=”Mobile number is required.”)nn new_otp = await otp_service.generate_and_send_otp(db, mobile=mobile)n if not new_otp:n raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”No pending registration found for this mobile number.”)nn return {“message”: “A new OTP has been sent to your mobile number.”}n”
},
{
“path”: “src/schemas/user_schemas.py”,
“code_template”: “from pydantic import BaseModel, EmailStr, Field, validatornfrom typing import Optionalnimport renn# Pydantic schema for data validation and serializationnnclass UserBase(BaseModel):n email: EmailStrn mobile: strnn @validator(‘mobile’)n def validate_mobile(cls, v):n # Example validation for a 10-digit mobile numbern if not re.match(r’^\d{10}$’, v):n raise ValueError(‘Mobile number must be 10 digits.’)n return vnnclass UserCreate(UserBase):n password: str = Field(…, min_length=8)nnclass UserResponse(UserBase):n id: intn is_active: boolnn class Config:n orm_mode = Truennclass OTPVerifyRequest(BaseModel):n mobile: strn otp: str = Field(…, min_length=6, max_length=6)nn @validator(‘mobile’)n def validate_mobile(cls, v):n if not re.match(r’^\d{10}$’, v):n raise ValueError(‘Mobile number must be 10 digits.’)n return vn”
},{
“path”: “src/services/user_service.py”,
“code_template”: “from sqlalchemy.orm import Sessionnfrom fastapi import HTTPException, statusnnfrom src.models.user_model import Usernfrom src.schemas.user_schemas import UserCreatenfrom src.core.security import get_password_hashnfrom src.services import otp_servicenndef get_user_by_email_or_mobile(db: Session, email: str, mobile: str):n “””Retrieve a user by either email or mobile number.”””n return db.query(User).filter((User.email == email) | (User.mobile == mobile)).first()nnasync def create_inactive_user_and_send_otp(db: Session, user_in: UserCreate) -> str:n “””n Creates a temporary, inactive user and sends an OTP.n If an inactive user already exists, it updates their details and resends OTP.n “””n existing_user = db.query(User).filter(User.mobile == user_in.mobile).first()nn if existing_user and existing_user.is_active:n raise HTTPException(n status_code=status.HTTP_409_CONFLICT,n detail=”An active account with this mobile number already exists.”n )nn if existing_user:n # Update existing inactive user’s infon existing_user.email = user_in.emailn # Store hashed password temporarily, or handle differently based on security policyn existing_user.hashed_password = get_password_hash(user_in.password)n else:n # Create new inactive usern db_user = User(n email=user_in.email,n mobile=user_in.mobile,n hashed_password=get_password_hash(user_in.password), # Hash password before storingn is_active=False # User is not active until OTP is verifiedn )n db.add(db_user)n n db.commit()nn # Generate and send OTPn otp_code = await otp_service.generate_and_send_otp(db, mobile=user_in.mobile)n return otp_codennndef activate_user(db: Session, mobile: str) -> User:n “””n Activates a user account after successful OTP verification.n “””n user = db.query(User).filter(User.mobile == mobile, User.is_active == False).first()n if user:n user.is_active = Truen db.commit()n db.refresh(user)n return usern”
},
{
“path”: “src/services/otp_service.py”,
“code_template”: “import randomnimport datetimenfrom sqlalchemy.orm import Sessionnnfrom src.models.otp_model import OTPnfrom src.core.config import settingsnn# This is a mock OTP sending function. In production, integrate with an SMS gateway like Twilio.nasync def _send_sms(mobile: str, message: str):n print(f”— MOCK SMS GATEWAY —“)n print(f”To: {mobile}”)n print(f”Message: {message}”)n print(f”————————“)n # Simulate network delayn # await asyncio.sleep(1)n return Truennasync def generate_and_send_otp(db: Session, mobile: str) -> str:n “””n Generates a 6-digit OTP, stores it in the database with an expiry, and sends it via SMS.n “””n # Invalidate any existing OTPs for this numbern db.query(OTP).filter(OTP.mobile == mobile).delete()nn otp_code = str(random.randint(100000, 999999))n expiry_time = datetime.datetime.utcnow() + datetime.timedelta(minutes=settings.OTP_EXPIRY_MINUTES)nn db_otp = OTP(mobile=mobile, otp_code=otp_code, expires_at=expiry_time)n db.add(db_otp)n db.commit()nn message = f”Your verification code for Great World is: {otp_code}. It is valid for {settings.OTP_EXPIRY_MINUTES} minutes.”n await _send_sms(mobile, message)nn return otp_codennasync def verify_otp(db: Session, mobile: str, otp_code: str) -> bool:n “””n Verifies if the provided OTP is correct and not expired.n Deletes the OTP after verification to prevent reuse.n “””n db_otp = db.query(OTP).filter(OTP.mobile == mobile, OTP.otp_code == otp_code).first()nn if not db_otp:n return Falsenn is_expired = db_otp.expires_at < datetime.datetime.utcnow()n if is_expired:n db.delete(db_otp)n db.commit()n return Falsenn # OTP is valid, delete it and return Truen db.delete(db_otp)n db.commit()n return Truen”
},
{
“path”: “src/models/user_model.py”,
“code_template”: “from sqlalchemy import Column, Integer, String, Booleannfrom src.db.base_class import Basennclass User(Base):n __tablename__ = “users”nn id = Column(Integer, primary_key=True, index=True)n email = Column(String, unique=True, index=True, nullable=False)n mobile = Column(String, unique=True, index=True, nullable=False)n hashed_password = Column(String, nullable=False)n is_active = Column(Boolean, default=False)n”
},
{
“path”: “src/models/otp_model.py”,
“code_template”: “from sqlalchemy import Column, Integer, String, DateTime, ForeignKeynfrom src.db.base_class import Basennclass OTP(Base):n __tablename__ = “otps”nn id = Column(Integer, primary_key=True, index=True)n mobile = Column(String, index=True, nullable=False)n otp_code = Column(String, nullable=False)n expires_at = Column(DateTime, nullable=False)n”
},
{
“path”: “src/core/config.py”,
“code_template”: “from pydantic import BaseSettingsnnclass Settings(BaseSettings):n # In a real app, these would come from environment variablesn DATABASE_URL: str = “sqlite:///./great_world.db”n SECRET_KEY: str = “a_very_secret_key_for_jwt”n ALGORITHM: str = “HS256″n ACCESS_TOKEN_EXPIRE_MINUTES: int = 30n OTP_EXPIRY_MINUTES: int = 5nn class Config:n env_file = “.env”nnsettings = Settings()n”
}
]
}
}