I built an application with Fastapi+MongoDB, the following code about it, when I tested it, I found that Fastapi’s docs don’t parse the send-code and verify-code interfaces, I searched for a lot of ways to do it but couldn’t solve it, what’s going on?
main.py:
@app.post("/send-code/")
async def send_code(phone_verification: PhoneVerification):
if db.users.find_one({"phone_number": phone_verification.phone_number, "is_blacklisted": True}):
raise HTTPException(status_code=400, detail="Phone number is blacklisted.")
existing = db.phone_verifications.find_one({"phone_number": phone_verification.phone_number})
now = datetime.utcnow()
if existing:
if existing['attempts'] >= 2:
raise HTTPException(status_code=400, detail="Maximum request limit reached for today.")
if (now - existing['last_attempt']).total_seconds() < 18000:
raise HTTPException(status_code=400, detail="Request too soon. Wait for at least 5 hours.")
phone_verification.attempts = (existing['attempts'] + 1) if existing else 1
phone_verification.last_attempt = now
phone_verification.created_at = now if not existing else existing['created_at']
db.phone_verifications.update_one(
{"phone_number": phone_verification.phone_number},
{"$set": {
"attempts": phone_verification.attempts,
"last_attempt": phone_verification.last_attempt,
"code": phone_verification.code,
"created_at": phone_verification.created_at
}},
upsert=True
)
return {"message": "Verification code sent."}
@app.post("/verify-code/")
async def verify_code(phone_number: str, code: str):
verification = db.phone_verifications.find_one({"phone_number": phone_number})
if not verification or (datetime.utcnow() - verification['created_at']).total_seconds() > 3600:
raise HTTPException(status_code=400, detail="Code is invalid or expired.")
if verification['code'] == code:
db.users.update_one({"phone_number": phone_number}, {"$set": {"phone_verified": True}})
return {"message": "Phone number verified successfully."}
else:
raise HTTPException(status_code=400, detail="Invalid verification code.")
models.py:
class PhoneVerification(BaseModel):
phone_number: str
code: str = Field(default_factory=lambda: ''.join([str(randint(0, 9)) for _ in range(6)]))
created_at: datetime = Field(default_factory=datetime.utcnow)
attempts: int = 0
last_attempt: datetime = Field(default_factory=datetime.utcnow)
@validator('phone_number')
def validate_phone_number(cls, v):
pattern = re.compile(r'^1(3[0-9]|4[01456879]|5[0-35-9]|6[2567]|7[0-8]|8[0-9]|9[0-35-9])d{8}$')
if not pattern.match(v):
raise ValueError("Invalid phone number format")
return v
I tried to send it a request by sending it a request and it returned No Found.Here is the code for its minimal instantiation file:
import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
from datetime import datetime
from pymongo import MongoClient
from random import randint
import re
app = FastAPI()
client = MongoClient("mongodb://localhost:27017/")
db = client.your_database
# 配置日志记录
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PhoneVerification(BaseModel):
phone_number: str
code: str = Field(default_factory=lambda: ''.join([str(randint(0, 9)) for _ in range(6)]))
created_at: datetime = Field(default_factory=datetime.utcnow)
attempts: int = 0
last_attempt: datetime = Field(default_factory=datetime.utcnow)
@validator('phone_number')
def validate_phone_number(cls, v):
pattern = re.compile(r'^1(3[0-9]|4[01456879]|5[0-35-9]|6[2567]|7[0-8]|8[0-9]|9[0-35-9])d{8}$')
if not pattern.match(v):
raise ValueError("Invalid phone number format")
return v
@app.post("/send-code/")
async def send_code(phone_verification: PhoneVerification):
logger.info(f"Received request to send code to {phone_verification.phone_number}")
if db.users.find_one({"phone_number": phone_verification.phone_number, "is_blacklisted": True}):
logger.warning(f"Phone number {phone_verification.phone_number} is blacklisted")
raise HTTPException(status_code=400, detail="Phone number is blacklisted.")
existing = db.phone_verifications.find_one({"phone_number": phone_verification.phone_number})
now = datetime.utcnow()
if existing:
if existing['attempts'] >= 2:
logger.warning(f"Maximum request limit reached for {phone_verification.phone_number}")
raise HTTPException(status_code=400, detail="Maximum request limit reached for today.")
if (now - existing['last_attempt']).total_seconds() < 18000:
logger.warning(f"Request too soon for {phone_verification.phone_number}")
raise HTTPException(status_code=400, detail="Request too soon. Wait for at least 5 hours.")
phone_verification.attempts = (existing['attempts'] + 1) if existing else 1
phone_verification.last_attempt = now
phone_verification.created_at = now if not existing else existing['created_at']
db.phone_verifications.update_one(
{"phone_number": phone_verification.phone_number},
{"$set": {
"attempts": phone_verification.attempts,
"last_attempt": phone_verification.last_attempt,
"code": phone_verification.code,
"created_at": phone_verification.created_at
}},
upsert=True
)
logger.info(f"Verification code sent to {phone_verification.phone_number}")
return {"message": "Verification code sent."}
@app.post("/verify-code/")
async def verify_code(phone_number: str, code: str):
logger.info(f"Received request to verify code for {phone_number}")
verification = db.phone_verifications.find_one({"phone_number": phone_number})
if not verification or (datetime.utcnow() - verification['created_at']).total_seconds() > 3600:
logger.warning(f"Code is invalid or expired for {phone_number}")
raise HTTPException(status_code=400, detail="Code is invalid or expired.")
if verification['code'] == code:
db.users.update_one({"phone_number": phone_number}, {"$set": {"phone_verified": True}})
logger.info(f"Phone number {phone_number} verified successfully")
return {"message": "Phone number verified successfully."}
else:
logger.warning(f"Invalid verification code for {phone_number}")
raise HTTPException(status_code=400, detail="Invalid verification code.")
hezizizi is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.