restapi with python, fastapi and mongodb

Author: wpcodevo

.env

@@ -0,0 +1,15 @@
+MONGO_INITDB_ROOT_USERNAME=admin
+MONGO_INITDB_ROOT_PASSWORD=password123
+MONGO_INITDB_DATABASE=fastapi
+
+DATABASE_URL=mongodb://admin:password123@localhost:6000/fastapi?authSource=admin
+
+ACCESS_TOKEN_EXPIRES_IN=15
+REFRESH_TOKEN_EXPIRES_IN=60
+JWT_ALGORITHM=RS256
+
+CLIENT_ORIGIN=http://localhost:3000
+
+
+JWT_PRIVATE_KEY=LS0tLS1CRUdJTiBSU0EgUFJJVkFURSBLRVktLS0tLQpNSUlCT2dJQkFBSkJBSSs3QnZUS0FWdHVQYzEzbEFkVk94TlVmcWxzMm1SVmlQWlJyVFpjd3l4RVhVRGpNaFZuCi9KVHRsd3h2a281T0pBQ1k3dVE0T09wODdiM3NOU3ZNd2xNQ0F3RUFBUUpBYm5LaENOQ0dOSFZGaHJPQ0RCU0IKdmZ2ckRWUzVpZXAwd2h2SGlBUEdjeWV6bjd0U2RweUZ0NEU0QTNXT3VQOXhqenNjTFZyb1pzRmVMUWlqT1JhUwp3UUloQU84MWl2b21iVGhjRkltTFZPbU16Vk52TGxWTW02WE5iS3B4bGh4TlpUTmhBaUVBbWRISlpGM3haWFE0Cm15QnNCeEhLQ3JqOTF6bVFxU0E4bHUvT1ZNTDNSak1DSVFEbDJxOUdtN0lMbS85b0EyaCtXdnZabGxZUlJPR3oKT21lV2lEclR5MUxaUVFJZ2ZGYUlaUWxMU0tkWjJvdXF4MHdwOWVEejBEWklLVzVWaSt6czdMZHRDdUVDSUVGYwo3d21VZ3pPblpzbnU1clBsTDJjZldLTGhFbWwrUVFzOCtkMFBGdXlnCi0tLS0tRU5EIFJTQSBQUklWQVRFIEtFWS0tLS0t
+JWT_PUBLIC_KEY=LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUZ3d0RRWUpLb1pJaHZjTkFRRUJCUUFEU3dBd1NBSkJBSSs3QnZUS0FWdHVQYzEzbEFkVk94TlVmcWxzMm1SVgppUFpSclRaY3d5eEVYVURqTWhWbi9KVHRsd3h2a281T0pBQ1k3dVE0T09wODdiM3NOU3ZNd2xNQ0F3RUFBUT09Ci0tLS0tRU5EIFBVQkxJQyBLRVktLS0tLQ==

.gitignore

@@ -0,0 +1,3 @@
+__pycache__
+venv/
+# .env

Makefile

@@ -0,0 +1,5 @@
+dev:
+	docker-compose up -d
+
+dev-down:
+	docker-compose down

app/__init__.py

@@ -0,0 +1,20 @@
+from pydantic import BaseSettings
+
+
+class Settings(BaseSettings):
+    DATABASE_URL: str
+    MONGO_INITDB_DATABASE: str
+
+    JWT_PUBLIC_KEY: str
+    JWT_PRIVATE_KEY: str
+    REFRESH_TOKEN_EXPIRES_IN: int
+    ACCESS_TOKEN_EXPIRES_IN: int
+    JWT_ALGORITHM: str
+
+    CLIENT_ORIGIN: str
+
+    class Config:
+        env_file = './.env'
+
+
+settings = Settings()

app/config.py

@@ -0,0 +1,10 @@
+from pymongo import mongo_client
+import pymongo
+from app.config import settings
+
+client = mongo_client.MongoClient(settings.DATABASE_URL)
+print('Connected to MongoDB...')
+
+db = client[settings.MONGO_INITDB_DATABASE]
+User = db.users
+User.create_index([("email", pymongo.ASCENDING)], unique=True)

app/database.py

@@ -0,0 +1,28 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.config import settings
+from app.routers import auth, user
+
+app = FastAPI()
+
+origins = [
+    settings.CLIENT_ORIGIN,
+]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+app.include_router(auth.router, tags=['Auth'], prefix='/api/auth')
+app.include_router(user.router, tags=['Users'], prefix='/api/users')
+
+
+@app.get("/api/healthchecker")
+def root():
+    return {"message": "Welcome to FastAPI with MongoDB"}

app/main.py

@@ -0,0 +1,66 @@
+import base64
+from typing import List
+from fastapi import Depends, HTTPException, status
+from fastapi_jwt_auth import AuthJWT
+from pydantic import BaseModel
+from bson.objectid import ObjectId
+
+from app.serializers import userEntity
+
+from .database import User
+from .config import settings
+
+
+class Settings(BaseModel):
+    authjwt_algorithm: str = settings.JWT_ALGORITHM
+    authjwt_decode_algorithms: List[str] = [settings.JWT_ALGORITHM]
+    authjwt_token_location: set = {'cookies', 'headers'}
+    authjwt_access_cookie_key: str = 'access_token'
+    authjwt_refresh_cookie_key: str = 'refresh_token'
+    authjwt_cookie_csrf_protect: bool = False
+    authjwt_public_key: str = base64.b64decode(
+        settings.JWT_PUBLIC_KEY).decode('utf-8')
+    authjwt_private_key: str = base64.b64decode(
+        settings.JWT_PRIVATE_KEY).decode('utf-8')
+
+
+@AuthJWT.load_config
+def get_config():
+    return Settings()
+
+
+class NotVerified(Exception):
+    pass
+
+
+class UserNotFound(Exception):
+    pass
+
+
+def require_user(Authorize: AuthJWT = Depends()):
+    try:
+        Authorize.jwt_required()
+        user_id = Authorize.get_jwt_subject()
+        user = userEntity(User.find_one({'_id': ObjectId(str(user_id))}))
+
+        if not user:
+            raise UserNotFound('User no longer exist')
+
+        if not user["verified"]:
+            raise NotVerified('You are not verified')
+
+    except Exception as e:
+        error = e.__class__.__name__
+        print(error)
+        if error == 'MissingTokenError':
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail='You are not logged in')
+        if error == 'UserNotFound':
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail='User no longer exist')
+        if error == 'NotVerified':
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED, detail='Please verify your account')
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED, detail='Token is invalid or has expired')
+    return user_id

app/oauth2.py

@@ -0,0 +1,116 @@
+from datetime import datetime, timedelta
+from bson.objectid import ObjectId
+from fastapi import APIRouter, Request, Response, status, Depends, HTTPException
+from pydantic import EmailStr
+
+from app import oauth2
+from app.database import User
+from app.serializers import userEntity, userResponseEntity
+from .. import schemas, utils
+from app.oauth2 import AuthJWT
+from ..config import settings
+
+
+router = APIRouter()
+ACCESS_TOKEN_EXPIRES_IN = settings.ACCESS_TOKEN_EXPIRES_IN
+REFRESH_TOKEN_EXPIRES_IN = settings.REFRESH_TOKEN_EXPIRES_IN
+
+
+@router.post('/register', status_code=status.HTTP_201_CREATED, response_model=schemas.UserResponse)
+async def create_user(payload: schemas.CreateUserSchema):
+    # Check if user already exist
+    user = User.find_one({'email': payload.email.lower()})
+    if user:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT,
+                            detail='Account already exist')
+    # Compare password and passwordConfirm
+    if payload.password != payload.passwordConfirm:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail='Passwords do not match')
+    #  Hash the password
+    payload.password = utils.hash_password(payload.password)
+    del payload.passwordConfirm
+    payload.role = 'user'
+    payload.verified = True
+    payload.email = payload.email.lower()
+    payload.created_at = datetime.utcnow()
+    payload.updated_at = payload.created_at
+    result = User.insert_one(payload.dict())
+    new_user = userResponseEntity(User.find_one({'_id': result.inserted_id}))
+    return {"status": "success", "user": new_user}
+
+
+@router.post('/login')
+def login(payload: schemas.LoginUserSchema, response: Response, Authorize: AuthJWT = Depends()):
+    # Check if the user exist
+    user = userEntity(User.find_one({'email': payload.email.lower()}))
+    if not user:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
+                            detail='Incorrect Email or Password')
+
+    # Check if user verified his email
+    if not user['verified']:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                            detail='Please verify your email address')
+
+    # Check if the password is valid
+    if not utils.verify_password(payload.password, user['password']):
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
+                            detail='Incorrect Email or Password')
+
+    # Create access token
+    access_token = Authorize.create_access_token(
+        subject=str(user["id"]), expires_time=timedelta(minutes=ACCESS_TOKEN_EXPIRES_IN))
+
+    # Create refresh token
+    refresh_token = Authorize.create_refresh_token(
+        subject=str(user["id"]), expires_time=timedelta(minutes=REFRESH_TOKEN_EXPIRES_IN))
+
+    # Store refresh and access tokens in cookie
+    response.set_cookie('access_token', access_token, ACCESS_TOKEN_EXPIRES_IN * 60,
+                        ACCESS_TOKEN_EXPIRES_IN * 60, '/', None, False, True, 'lax')
+    response.set_cookie('refresh_token', refresh_token,
+                        REFRESH_TOKEN_EXPIRES_IN * 60, REFRESH_TOKEN_EXPIRES_IN * 60, '/', None, False, True, 'lax')
+    response.set_cookie('logged_in', 'True', ACCESS_TOKEN_EXPIRES_IN * 60,
+                        ACCESS_TOKEN_EXPIRES_IN * 60, '/', None, False, False, 'lax')
+
+    # Send both access
+    return {'status': 'success', 'access_token': access_token}
+
+
+@router.get('/refresh')
+def refresh_token(response: Response, Authorize: AuthJWT = Depends()):
+    try:
+        Authorize.jwt_refresh_token_required()
+
+        user_id = Authorize.get_jwt_subject()
+        if not user_id:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                                detail='Could not refresh access token')
+        user = userEntity(User.find_one({'_id': ObjectId(str(user_id))}))
+        if not user:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,
+                                detail='The user belonging to this token no logger exist')
+        access_token = Authorize.create_access_token(
+            subject=str(user["id"]), expires_time=timedelta(minutes=ACCESS_TOKEN_EXPIRES_IN))
+    except Exception as e:
+        error = e.__class__.__name__
+        if error == 'MissingTokenError':
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST, detail='Please provide refresh token')
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail=error)
+
+    response.set_cookie('access_token', access_token, ACCESS_TOKEN_EXPIRES_IN * 60,
+                        ACCESS_TOKEN_EXPIRES_IN * 60, '/', None, False, True, 'lax')
+    response.set_cookie('logged_in', 'True', ACCESS_TOKEN_EXPIRES_IN * 60,
+                        ACCESS_TOKEN_EXPIRES_IN * 60, '/', None, False, False, 'lax')
+    return {'access_token': access_token}
+
+
+@router.get('/logout', status_code=status.HTTP_200_OK)
+def logout(response: Response, Authorize: AuthJWT = Depends(), user_id: str = Depends(oauth2.require_user)):
+    Authorize.unset_jwt_cookies()
+    response.set_cookie('logged_in', '', -1)
+
+    return {'status': 'success'}

app/routers/auth.py

@@ -0,0 +1,14 @@
+from fastapi import APIRouter, Depends
+from bson.objectid import ObjectId
+from app.serializers import userResponseEntity
+
+from app.database import User
+from .. import schemas, oauth2
+
+router = APIRouter()
+
+
+@router.get('/me', response_model=schemas.UserResponse)
+def get_me(user_id: str = Depends(oauth2.require_user)):
+    user = userResponseEntity(User.find_one({'_id': ObjectId(str(user_id))}))
+    return {"status": "success", "user": user}

app/routers/user.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+from pydantic import BaseModel, EmailStr, constr
+
+
+class UserBaseSchema(BaseModel):
+    id: str | None = None
+    name: str
+    email: str
+    photo: str
+    role: str | None = None
+    created_at: datetime | None = None
+    updated_at: datetime | None = None
+
+    class Config:
+        orm_mode = True
+
+
+class CreateUserSchema(UserBaseSchema):
+    password: constr(min_length=8)
+    passwordConfirm: str
+    verified: bool = False
+
+
+class LoginUserSchema(BaseModel):
+    email: EmailStr
+    password: constr(min_length=8)
+
+
+class UserResponse(BaseModel):
+    status: str
+    user: UserBaseSchema

app/schemas.py

@@ -0,0 +1,28 @@
+def userEntity(user) -> dict:
+    return {
+        "id": str(user["_id"]),
+        "name": user["name"],
+        "email": user["email"],
+        "role": user["role"],
+        "photo": user["photo"],
+        "verified": user["verified"],
+        "password": user["password"],
+        "created_at": user["created_at"],
+        "updated_at": user["updated_at"]
+    }
+
+
+def userResponseEntity(user) -> dict:
+    return {
+        "id": str(user["_id"]),
+        "name": user["name"],
+        "email": user["email"],
+        "role": user["role"],
+        "photo": user["photo"],
+        "created_at": user["created_at"],
+        "updated_at": user["updated_at"]
+    }
+
+
+def userListEntity(users) -> list:
+    return [userEntity(user) for user in users]

app/serializers.py

@@ -0,0 +1,11 @@
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(password: str):
+    return pwd_context.hash(password)
+
+
+def verify_password(password: str, hashed_password: str):
+    return pwd_context.verify(password, hashed_password)

app/utils.py

@@ -0,0 +1,18 @@
+version: '3'
+services:
+  mongo:
+    image: mongo:latest
+    container_name: mongo
+    env_file:
+      - ./.env
+    environment:
+      MONGO_INITDB_ROOT_USERNAME: ${MONGO_INITDB_ROOT_USERNAME}
+      MONGO_INITDB_ROOT_PASSWORD: ${MONGO_INITDB_ROOT_PASSWORD}
+      MONGO_INITDB_DATABASE: ${MONGO_INITDB_DATABASE}
+    volumes:
+      - mongo:/data/db
+    ports:
+      - '6000:27017'
+
+volumes:
+  mongo: