FastAPI SMS 전송 및 인증 확인 + 로그인 구현 with coolsms

requirements.py 

annotated-types==0.7.0
anyio==4.4.0
certifi==2024.7.4
charset-normalizer==3.3.2
click==8.1.7
coolsms==0.4
coolsms_python_sdk==2.0.3
fastapi==0.112.0
h11==0.14.0
idna==3.7
pydantic==2.8.2
pydantic_core==2.20.1
PyMySQL==1.1.1
python-dotenv==1.0.1
requests==2.32.3
sniffio==1.3.1
SQLAlchemy==2.0.31
starlette==0.37.2
typing_extensions==4.12.2
urllib3==2.2.2
uvicorn==0.30.5

 

작성한 후 아래 명령어를 터미널에 입력하면 필요한 패키지들을 자동으로 설치해 준다.

pip install -r requirements.txt

 

main.py 

import jwt
from datetime import datetime, timedelta
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from sqlalchemy.orm import Session
import random
import string
from sdk.api.message import Message
from sdk.exceptions import CoolsmsException
from database import SessionLocal, Base, engine
from models import User, AuthPhone, AuthToken
from dotenv import load_dotenv
import os
import logging
import pytz

# .env 파일의 환경 변수를 로드합니다.
load_dotenv()

# 환경 변수에서 API_KEY, API_SECRET, SENDER_NUMBER, JWT_SECRET_KEY, JWT_ALGORITHM를 가져옵니다.
SMS_API_KEY = os.getenv("SMS_API_KEY")
SMS_API_SECRET = os.getenv("SMS_API_SECRET")
SENDER_NUMBER = os.getenv("SENDER_NUMBER")
JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY")
JWT_ALGORITHM = os.getenv("JWT_ALGORITHM")

# FastAPI 애플리케이션 인스턴스를 생성합니다.
app = FastAPI()

# 데이터베이스 테이블 생성 (필요한 경우에만 사용)
Base.metadata.create_all(bind=engine)

# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# 서울 시간대 설정
KST = pytz.timezone("Asia/Seoul")


# 요청 데이터 모델 정의
class PhoneNumberRequest(BaseModel):
    phone_number: str


class VerifyCodeRequest(BaseModel):
    phone_number: str
    auth_code: str


class UserSignupRequest(BaseModel):
    student_number: str
    name: str
    email: str
    contact: str
    department: str


# 인증 코드를 생성하는 함수
def generate_verification_code(length=6):
    return "".join(random.choices(string.digits, k=length))


# JWT 액세스 토큰을 생성하는 함수
def create_access_token(student_number: str):
    expire = datetime.utcnow() + timedelta(hours=1)
    payload = {"sub": student_number, "exp": expire}
    token = jwt.encode(payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
    return token


# 데이터베이스 세션을 가져오는 종속성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# SMS를 전송하는 엔드포인트
@app.post("/send_sms/")
def send_sms(request: PhoneNumberRequest, db: Session = Depends(get_db)):
    verification_code = generate_verification_code()
    expires_at_kst = datetime.now(tz=KST) + timedelta(minutes=5)

    # 인증 정보를 데이터베이스에 저장
    auth_phone = AuthPhone(
        phone_number=request.phone_number,
        auth_code=verification_code,
        expires_at=expires_at_kst,
    )

    db.add(auth_phone)
    db.commit()

    # SMS 전송 파라미터 설정
    params = {
        "type": "sms",
        "to": request.phone_number,
        "from": SENDER_NUMBER,
        "text": f"인증번호는 {verification_code}입니다.",
    }
    cool = Message(SMS_API_KEY, SMS_API_SECRET)
    try:
        response = cool.send(params)
        if response["success_count"] > 0:
            logger.info(f"SMS sent successfully to {request.phone_number}")
            return {"message": "SMS sent successfully"}
        else:
            logger.error(f"Failed to send SMS to {request.phone_number}")
            raise HTTPException(status_code=400, detail="Failed to send SMS")

    except CoolsmsException as e:
        logger.error(f"Internal Server Error: {e.msg}")
        raise HTTPException(status_code=500, detail=f"Internal Server Error: {e.msg}")


# 인증 코드를 검증하는 엔드포인트
@app.post("/verify_sms/")
def verify_sms(request: VerifyCodeRequest, db: Session = Depends(get_db)):
    current_time_kst = datetime.now(tz=KST)
    auth_phone = (
        db.query(AuthPhone)
        .filter(
            AuthPhone.phone_number == request.phone_number,
            AuthPhone.auth_code == request.auth_code,
            AuthPhone.expires_at > current_time_kst,
            AuthPhone.is_verified == False,
        )
        .first()
    )

    if not auth_phone:
        logger.warning(
            f"Invalid or expired verification code for {request.phone_number}"
        )
        raise HTTPException(
            status_code=400, detail="Invalid or expired verification code"
        )

    auth_phone.is_verified = True
    db.commit()

    user = db.query(User).filter(User.contact == request.phone_number).first()

    if user:
        token = create_access_token(user.student_number)
        auth_token = AuthToken(
            student_number=user.student_number,
            token_type="access",
            token=token,
            created_at=current_time_kst,
            expires_at=current_time_kst + timedelta(hours=1),
        )
        db.add(auth_token)
        db.commit()
        logger.info(f"Login successful for {user.student_number}")
        return {"message": "Login successful", "token": token}
    else:
        logger.info(
            f"Verification successful for {request.phone_number}, proceed to signup"
        )
        return {"message": "Verification successful, proceed to signup"}


# 사용자 등록을 처리하는 엔드포인트
@app.post("/signup/")
def signup(request: UserSignupRequest, db: Session = Depends(get_db)):
    signup_date_kst = datetime.now(tz=KST)
    user = User(
        student_number=request.student_number,
        name=request.name,
        email=request.email,
        contact=request.contact,
        department=request.department,
        signup_date=signup_date_kst,
    )
    db.add(user)
    db.commit()

    token = create_access_token(user.student_number)
    auth_token = AuthToken(
        student_number=user.student_number,
        token_type="access",
        token=token,
        created_at=signup_date_kst,
        expires_at=signup_date_kst + timedelta(hours=1),
    )
    db.add(auth_token)
    db.commit()

    logger.info(f"Signup successful for {request.student_number}")
    return {"message": "Signup successful", "token": token}


# uvicorn을 사용하여 애플리케이션을 실행합니다.
if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

 

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from dotenv import load_dotenv
import os

# .env 파일의 환경 변수를 로드합니다.
load_dotenv()

# 환경 변수에서 DATABASE_URL을 가져옵니다.
DATABASE_URL = os.getenv("DATABASE_URL")

# 데이터베이스 엔진 생성
engine = create_engine(DATABASE_URL)

# 세션 로컬 클래스 생성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 베이스 클래스 생성
Base = declarative_base()

 

models.py

# models.py
from sqlalchemy import Column, Integer, String, Boolean, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from database import Base
from datetime import datetime


class User(Base):
    __tablename__ = "User"
    id = Column(Integer, primary_key=True, index=True)
    student_number = Column(String(20), unique=True, index=True, nullable=False)
    name = Column(String(100), nullable=False)
    email = Column(String(100), nullable=False)
    contact = Column(String(20))
    request_details = Column(String(200))
    department = Column(String(100))
    signup_date = Column(DateTime, default=datetime.utcnow)
    deactivation_date = Column(DateTime)


class AuthPhone(Base):
    __tablename__ = "auth_phone"
    auth_phone_id = Column(Integer, primary_key=True, index=True)
    phone_number = Column(String(20), index=True)
    auth_code = Column(String(6), index=True)
    is_verified = Column(Boolean, default=False)
    expires_at = Column(DateTime)


class AuthToken(Base):
    __tablename__ = "auth_token"
    token_id = Column(Integer, primary_key=True, index=True)
    student_number = Column(String(20), ForeignKey("User.student_number"), index=True)
    token_type = Column(String(255))
    token = Column(String(512))  # Increase length to 512 if necessary
    created_at = Column(DateTime, default=datetime.utcnow)
    expires_at = Column(DateTime)

 

init_db.py(테스트 중 테이블 초기화 시 사용)

from sqlalchemy import create_engine
from database import Base
from models import User, AuthPhone, AuthToken
from dotenv import load_dotenv
import os

# .env 파일의 환경 변수를 로드합니다.
load_dotenv()

# 환경 변수에서 DATABASE_URL을 가져옵니다.
DATABASE_URL = os.getenv("DATABASE_URL")

# 데이터베이스 엔진 생성
engine = create_engine(DATABASE_URL)


def reset_database():
    # 기존 테이블 삭제
    Base.metadata.drop_all(bind=engine)
    # 테이블 생성
    Base.metadata.create_all(bind=engine)
    print("Database has been reset and initialized.")


if __name__ == "__main__":
    reset_database()

 

.env

API_KEY = "coolsms에서 받은 api 키"
API_SECRET = "coolsms에서 받은 api 시크릿 키"
SENDER_NUMBER = "핸드폰 번호"
DATABASE_URL= "데이터베이스 주소"
JWT_SECRET_KEY="사용할 JWT 시크릿 키"
JWT_ALGORITHM=HS256

 

프로젝트 디렉토리 구조

project_directory/
│
├── main.py
├── database.py
├── init_db.py
├── models.py
├── .env
└── requirements.txt

 

SMS 전송 Curl문 예시

# SMS 전송 요청
curl -X 'POST' \
  'http://127.0.0.1:8000/send_code/' \
  -H 'Content-Type: application/json' \
  -d '{
  "phone_number": "01012345678"
}'

 

인증 코드 검증 Curl문 예시

# 인증 코드 검증 요청
curl -X 'POST' \
  'http://127.0.0.1:8000/verify_code/' \
  -H 'Content-Type: application/json' \
  -d '{
  "phone_number": "01012345678",
  "verification_code": "123456"
}'

 

회원가입 Curl문 예시

# 회원가입 요청
curl -X POST \
  http://127.0.0.1:8000/signup/ \
  -H "Content-Type: application/json" \
  -d '{
    "student_number": "2023123456",
    "name": "John Doe",
    "email": "johndoe@example.com",
    "contact": "01012345678",
    "department": "Computer Science"
  }'

 

실행 시 결과 값

SMS 전송을 성공했을 때

인증 코드를 다르게 입력했을 때

입력 코드를 올바르게 입력했을 때

회원가입에 성공했을 때