Đầu tiên là 1 vài lệnh ban đầu
sudo apt install python3.10-venv sudo python3 -m venv fastapi-mongodb/.venv source .venv/bin/activate pip3.10 install -r fastapi-mongodb/requirements.txt uvicorn main:app --reload
curl -X GET "http://localhost:8000/users" -H "accept: application/json"
code:
import motor.motor_asyncio
from fastapi import FastAPI, Response, Cookie, Request, status
from pydantic import BaseModel, EmailStr
from fastapi.encoders import jsonable_encoder
from bson import ObjectId
from pprint import pprint
from passlib.context import CryptContext
from fastapi.security import HTTPBasic, HTTPBasicCredentials
### CONSTANT
MONGODB_URL = "mongodb://127.0.0.1:27017/"
COOKIE_KEY = "0e7cdba83c30045082e0af5ff21c23180a862b7e32a53f2f5072ad606f95"
APP_SESSIONS = {}
### App
app = FastAPI()
security = HTTPBasic()
pwd_context = CryptContext(schemes=["bcrypt"])
### MongoDB
client = motor.motor_asyncio.AsyncIOMotorClient(MONGODB_URL)
db = client.itforvn
collection_users = db.get_collection("users")
class User(BaseModel):
fullname: str
email: EmailStr
password: str
class Config:
schema_extra = {
"example": {
"fullname": "John Doe",
"email": "john@gmail.com",
"password": "***",
}
}
class UserLogin(BaseModel):
email: EmailStr
password: str
def ResponseModel(data, message):
return {"data": [data], "code": 200, "message": message}
def ErrorResponseModel(error, code, message):
return {"error": error, "code": code, "message": message}
def user_helper(user) -> dict:
return {
"id": str(user["_id"]),
"fullname": user["fullname"],
"email": user["email"],
}
class Hasher:
@staticmethod
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
@staticmethod
def get_password_hash(password):
return pwd_context.hash(password)
@staticmethod
def get_cookie_hash(email):
return pwd_context.hash(COOKIE_KEY + email)
## User register - hashed password
@app.post("/users/register")
async def users_create(user: User, response: Response):
data = jsonable_encoder(user)
password_hash = Hasher.get_password_hash(data["password"])
data["password"] = password_hash
try:
add_user = await collection_users.insert_one(data)
new_user = await collection_users.find_one({"_id": add_user.inserted_id})
return ResponseModel(user_helper(new_user), "User added successfully.")
except:
return ErrorResponseModel("An error occurred.", 404, "User doesn't exist.")
# @app.post("/users/login")
# async def users_login(
# data: UserLogin,
# request: Request,
# response: Response,
# ):
# login_credentials = data.dict()
# _u_email = login_credentials["email"]
# _u_pass = login_credentials["password"]
# print(Hasher.get_password_hash(_u_pass))
# try:
# user = await collection_users.find_one({"email": _u_email})
# user_password = user["password"]
# print(f"hashed pass: {user_password}")
# if Hasher.verify_password(_u_pass, user_password):
# return {"message": "login success"}
# else:
# return {"message": "login fail"}
# except:
# return ErrorResponseModel("An error occurred.", 400, "Bad request!")
@app.get("/protected")
def users_protected(request: Request, response: Response):
try:
email = request.cookies.get("demo_email")
pprint(email)
login_cookie = request.cookies.get("demo_session")
if email and login_cookie and APP_SESSIONS[email] == login_cookie:
return {"message": "this is protected resource. enjoy!"}
else:
return ErrorResponseModel("An error occurred.", 403, "Request forbidden!")
except:
return ErrorResponseModel("An error occurred.", 403, "Request forbidden!")
@app.post("/users/login")
async def users_login(
data: UserLogin,
request: Request,
response: Response,
):
login_credential = data.dict()
_u_email = login_credential["email"]
_u_pass = login_credential["password"]
try:
user = await collection_users.find_one({"email": _u_email})
user_password = user["password"]
if Hasher.verify_password(_u_pass, user_password):
### Check if user logged in by cookie
try:
login_cookie = request.cookies.get("demo_session")
if login_cookie and APP_SESSIONS[_u_email] == login_cookie:
return {"message": "your are already login"}
except:
## Detect fraud
del APP_SESSIONS[_u_email]
return ErrorResponseModel("An error occurred.", 400, "Bad request!")
### If not logged in
cookie_hash = Hasher.get_cookie_hash(_u_email)
APP_SESSIONS[_u_email] = cookie_hash
response.set_cookie("demo_session", cookie_hash)
response.set_cookie("demo_email", _u_email)
return {"message": "login success"}
else:
return {"message": "login fail"}
except:
return ErrorResponseModel("An error occurred.", 400, "Bad request!")
@app.get("/users/logout")
def users_logout(request: Request, response: Response):
try:
email = request.cookies.get("demo_email")
del APP_SESSIONS[email]
response.delete_cookie("demo_email")
response.delete_cookie("demo_session")
return {"message": "goodbye"}
except:
return ErrorResponseModel("An error occurred.", 400, "Bad request!")
@app.get("/users")
async def user_list():
users = []
async for user in collection_users.find():
users.append(user_helper(user))
if users:
return ResponseModel(users, "Users data retrieved successfully")
return ResponseModel(users, "Empty list returned")
@app.get("/users/{id}")
async def user_get(id: str):
try:
user = await collection_users.find_one({"_id": ObjectId(id)})
except:
return ErrorResponseModel("An error occurred.", 400, "Invalid request")
if user:
return user_helper(user)
else:
return ErrorResponseModel("An error occurred.", 404, "User doesn't exist.")
class UpdateUser(BaseModel):
fullname: str | None = None
class Config:
schema_extra = {"example": {"fullname": "John Doe"}}
@app.put("/users/{id}")
async def user_update(id: str, req: UpdateUser):
data = {k: v for k, v in req.dict().items() if v is not None}
try:
user = await collection_users.find_one({"_id": ObjectId(id)})
if user:
updated_user = await collection_users.update_one(
{"_id": ObjectId(id)}, {"$set": data}
)
if updated_user:
return ResponseModel(
"User with ID: {} name update is successful".format(id),
"User updated successfully",
)
return ErrorResponseModel(
"An error occurred",
404,
"There was an error updating the user data.",
)
except:
return ErrorResponseModel("An error occurred.", 400, "Invalid request")
@app.delete("/users/{id}")
async def user_delete(id: str):
try:
user = await collection_users.find_one({"_id": ObjectId(id)})
if user:
await collection_users.delete_one({"_id": ObjectId(id)})
return ResponseModel(
"User with ID: {} removed".format(id), "User deleted successfully"
)
return ErrorResponseModel(
"An error occurred", 404, "User with id {0} doesn't exist".format(id)
)
except:
return ErrorResponseModel(
"An error occurred", 400, "User with id {0} is not valid".format(id)
)
file requirements.txt
anyio==3.6.2 black==22.10.0 bson==0.5.10 click==8.1.3 dnspython==2.2.1 email-validator==1.1.2 fastapi==0.88.0 gunicorn==20.1.0 h11==0.14.0 idna==3.4 more-itertools==8.14.0 motor==3.1.1 mypy-extensions==0.4.3 passlib==1.7.4 pathspec==0.10.2 platformdirs==2.5.4 pptree==3.1 pydantic==1.10.2 pymongo==4.3.3 python-dateutil==2.8.2 python-ulid==1.1.0 six==1.16.0 sniffio==1.3.0 starlette==0.22.0 tomli==2.0.1 types-redis==4.3.21.6 typing_extensions==4.4.0 uvicorn==0.20.0
