feat: add black & flake8 auto format config for vscode (#70)

* feat: add black & flake8 auto format config for vscode

* chore: update README.md for auto formatting by vscode extention

Author: qlawmarq

.vscode/extensions.json

@@ -0,0 +1,7 @@
+{
+  "recommendations": [
+    "ms-python.python",
+    "ms-python.flake8",
+    "ms-python.black-formatter"
+  ]
+}

.vscode/settings.json

@@ -0,0 +1,7 @@
+{
+  "editor.formatOnSave": true,
+  "[python]": {
+    "editor.defaultFormatter": "ms-python.black-formatter"
+  },
+  "flake8.args": ["--max-line-length=88", "--ignore=E203,W503,W504"]
+}

README.md

@@ -61,6 +61,19 @@ You can clone the front-end template from:
 
 ## Note
 
+### How to enable Python code formatter (black) and linter (flake8) with VSCode extension
+
+If you're [VS Code](https://code.visualstudio.com/) user, you can easily setup Python code formatter (black) and linter (flake8) by simply installing the extensions.
+
+Automatic formatting settings have already been defined here:
+
+`.vscode/settings.json`
+
+Just install following:
+
+- [Black Formatter](https://marketplace.visualstudio.com/items?itemName=ms-python.black-formatter)
+- [Flake8](https://marketplace.visualstudio.com/items?itemName=ms-python.flake8)
+
 ### How to check the DB tables in container
 
 You can check the DB data by actually executing a query using the following command:

api/database/query.py

@@ -1,21 +1,22 @@
-from pathlib import Path
 import os
 import pymysql.cursors
 from pymysql import converters
 
 
 converions = converters.conversions
-converions[pymysql.FIELD_TYPE.BIT] = lambda x: False if x == b'\x00' else True
+converions[pymysql.FIELD_TYPE.BIT] = lambda x: False if x == b"\x00" else True
 
 
 def init_connection():
-    connection = pymysql.connect(host=os.getenv("DATABASE_HOST"),
-                                 port=3306,
-                                 user=os.environ.get("DATABASE_USERNAME"),
-                                 password=os.environ.get("DATABASE_PASSWORD"),
-                                 database=os.environ.get("DATABASE"),
-                                 cursorclass=pymysql.cursors.DictCursor,
-                                 conv=converions)
+    connection = pymysql.connect(
+        host=os.getenv("DATABASE_HOST"),
+        port=3306,
+        user=os.environ.get("DATABASE_USERNAME"),
+        password=os.environ.get("DATABASE_PASSWORD"),
+        database=os.environ.get("DATABASE"),
+        cursorclass=pymysql.cursors.DictCursor,
+        conv=converions,
+    )
     return connection
 
 

api/main.py

@@ -1,10 +1,21 @@
-from fastapi import FastAPI, HTTPException, Security
+from fastapi import FastAPI, Security
 from fastapi.middleware.cors import CORSMiddleware
-from database.query import query_get, query_put, query_update
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from fastapi.responses import JSONResponse
 from fastapi.encoders import jsonable_encoder
-from user import Auth, SignInRequestModel, SignUpRequestModel, UserAuthResponseModel, UserUpdateRequestModel, UserResponseModel, register_user, signin_user, update_user, get_all_users, get_user_by_id
+from user import (
+    Auth,
+    SignInRequestModel,
+    SignUpRequestModel,
+    UserAuthResponseModel,
+    UserUpdateRequestModel,
+    UserResponseModel,
+    register_user,
+    signin_user,
+    update_user,
+    get_all_users,
+    get_user_by_id,
+)
 
 app = FastAPI()
 
@@ -14,7 +25,7 @@
     "http://localhost:3000",
     "http://localhost:3001",
     "http://localhost:4000",
-    "http://localhost:19006"
+    "http://localhost:19006",
 ]
 app.add_middleware(
     CORSMiddleware,
@@ -28,98 +39,124 @@
 auth_handler = Auth()
 
 
+"""
 ###############################
 ########## Auth APIs ##########
 ###############################
+"""
+
 
-@app.post('/v1/signup', response_model=UserAuthResponseModel)
+@app.post("/v1/signup", response_model=UserAuthResponseModel)
 def signup_api(user_details: SignUpRequestModel):
     """
     This sign-up API allow you to register your account, and return access token.
     """
     user = register_user(user_details)
     access_token = auth_handler.encode_token(user_details.email)
     refresh_token = auth_handler.encode_refresh_token(user_details.email)
-    return JSONResponse(status_code=200, content={'token': {'access_token': access_token, 'refresh_token': refresh_token}, 'user': user})
+    return JSONResponse(
+        status_code=200,
+        content={
+            "token": {"access_token": access_token, "refresh_token": refresh_token},
+            "user": user,
+        },
+    )
 
 
-@app.post('/v1/signin', response_model=UserAuthResponseModel)
+@app.post("/v1/signin", response_model=UserAuthResponseModel)
 def signin_api(user_details: SignInRequestModel):
     """
     This sign-in API allow you to obtain your access token.
     """
     user = signin_user(user_details.email, user_details.password)
-    access_token = auth_handler.encode_token(user['email'])
-    refresh_token = auth_handler.encode_refresh_token(user['email'])
-    return JSONResponse(status_code=200, content={'token': {'access_token': access_token, 'refresh_token': refresh_token}, 'user': user})
-
-
-@app.get('/v1/refresh-token')
+    access_token = auth_handler.encode_token(user["email"])
+    refresh_token = auth_handler.encode_refresh_token(user["email"])
+    return JSONResponse(
+        status_code=200,
+        content={
+            "token": {"access_token": access_token, "refresh_token": refresh_token},
+            "user": user,
+        },
+    )
+
+
+@app.get("/v1/refresh-token")
 def refresh_token_api(refresh_token: str):
     """
     This refresh-token API allow you to obtain new access token.
     """
     new_token = auth_handler.refresh_token(refresh_token)
-    return {'access_token': new_token}
+    return {"access_token": new_token}
 
 
+"""
 ################################
 ########## Users APIs ##########
 ################################
+"""
+
 
 @app.get("/v1/users", response_model=list[UserResponseModel])
 def get_all_users_api(credentials: HTTPAuthorizationCredentials = Security(security)):
     """
     This users get API allow you to fetch all user data.
     """
     token = credentials.credentials
-    if (auth_handler.decode_token(token)):
+    if auth_handler.decode_token(token):
         user = get_all_users()
         return JSONResponse(status_code=200, content=jsonable_encoder(user))
-    return JSONResponse(status_code=401, content={'error': 'Faild to authorize'})
+    return JSONResponse(status_code=401, content={"error": "Faild to authorize"})
 
 
 @app.get("/v1/user/{user_id}", response_model=UserResponseModel)
-def get_user_api(user_id: int, credentials: HTTPAuthorizationCredentials = Security(security)):
+def get_user_api(
+    user_id: int, credentials: HTTPAuthorizationCredentials = Security(security)
+):
     """
     This user API allow you to fetch specific user data.
     """
     token = credentials.credentials
-    if (auth_handler.decode_token(token)):
+    if auth_handler.decode_token(token):
         user = get_user_by_id(user_id)
         return JSONResponse(status_code=200, content=jsonable_encoder(user))
-    return JSONResponse(status_code=401, content={'error': 'Faild to authorize'})
+    return JSONResponse(status_code=401, content={"error": "Faild to authorize"})
 
 
 @app.post("/v1/user/update", response_model=UserResponseModel)
-def update_user_api(user_details: UserUpdateRequestModel, credentials: HTTPAuthorizationCredentials = Security(security)):
+def update_user_api(
+    user_details: UserUpdateRequestModel,
+    credentials: HTTPAuthorizationCredentials = Security(security),
+):
     """
     This user update API allow you to update user data.
     """
     token = credentials.credentials
-    if (auth_handler.decode_token(token)):
+    if auth_handler.decode_token(token):
         user = update_user(user_details)
         return JSONResponse(status_code=200, content=jsonable_encoder(user))
-    return JSONResponse(status_code=401, content={'error': 'Faild to authorize'})
+    return JSONResponse(status_code=401, content={"error": "Faild to authorize"})
 
 
+"""
 ###############################
 ########## Test APIs ##########
 ###############################
+"""
+
 
-@app.get('/secret')
+@app.get("/secret")
 def secret_data_api(credentials: HTTPAuthorizationCredentials = Security(security)):
     """
     This secret API is just for testing. Need access token to access this API.
     """
     token = credentials.credentials
-    if (auth_handler.decode_token(token)):
-        return 'Top Secret data only authorized users can access this info'
+    if auth_handler.decode_token(token):
+        return "Top Secret data only authorized users can access this info"
 
 
-@app.get('/not-secret')
+@app.get("/not-secret")
 def not_secret_data_api():
     """
     This not-secret API is just for testing.
     """
-    return 'Not secret data'
+    return "Not secret data"

api/user/auth.py

@@ -2,12 +2,13 @@
 import jwt  # used for encoding and decoding jwt tokens
 from fastapi import HTTPException  # used to handle error handling
 from passlib.context import CryptContext  # used for hashing the password
+
 # used to handle expiry time for tokens
 from datetime import datetime, timedelta
 
 
-class Auth():
-    hasher = CryptContext(schemes=['bcrypt'])
+class Auth:
+    hasher = CryptContext(schemes=["bcrypt"])
     secret = os.getenv("APP_SECRET_STRING")
 
     def encode_password(self, password):
@@ -18,55 +19,44 @@ def verify_password(self, password, encoded_password):
 
     def encode_token(self, email):
         payload = {
-            'exp': datetime.utcnow() + timedelta(days=0, minutes=30),
-            'iat': datetime.utcnow(),
-            'scope': 'access_token',
-            'sub': email
+            "exp": datetime.utcnow() + timedelta(days=0, minutes=30),
+            "iat": datetime.utcnow(),
+            "scope": "access_token",
+            "sub": email,
         }
-        return jwt.encode(
-            payload,
-            self.secret,
-            algorithm='HS256'
-        )
+        return jwt.encode(payload, self.secret, algorithm="HS256")
 
     def decode_token(self, token):
         try:
-            payload = jwt.decode(token, self.secret, algorithms=['HS256'])
-            if (payload['scope'] == 'access_token'):
-                return payload['sub']
+            payload = jwt.decode(token, self.secret, algorithms=["HS256"])
+            if payload["scope"] == "access_token":
+                return payload["sub"]
             raise HTTPException(
-                status_code=401, detail='Scope for the token is invalid')
+                status_code=401, detail="Scope for the token is invalid"
+            )
         except jwt.ExpiredSignatureError:
-            raise HTTPException(status_code=401, detail='Token expired')
+            raise HTTPException(status_code=401, detail="Token expired")
         except jwt.InvalidTokenError:
-            raise HTTPException(status_code=401, detail='Invalid token')
+            raise HTTPException(status_code=401, detail="Invalid token")
 
     def encode_refresh_token(self, email):
         payload = {
-            'exp': datetime.utcnow() + timedelta(days=0, hours=10),
-            'iat': datetime.utcnow(),
-            'scope': 'refresh_token',
-            'sub': email
+            "exp": datetime.utcnow() + timedelta(days=0, hours=10),
+            "iat": datetime.utcnow(),
+            "scope": "refresh_token",
+            "sub": email,
         }
-        return jwt.encode(
-            payload,
-            self.secret,
-            algorithm='HS256'
-        )
+        return jwt.encode(payload, self.secret, algorithm="HS256")
 
     def refresh_token(self, refresh_token):
         try:
-            payload = jwt.decode(
-                refresh_token, self.secret, algorithms=['HS256'])
-            if (payload['scope'] == 'refresh_token'):
-                email = payload['sub']
+            payload = jwt.decode(refresh_token, self.secret, algorithms=["HS256"])
+            if payload["scope"] == "refresh_token":
+                email = payload["sub"]
                 new_token = self.encode_token(email)
                 return new_token
-            raise HTTPException(
-                status_code=401, detail='Invalid scope for token')
+            raise HTTPException(status_code=401, detail="Invalid scope for token")
         except jwt.ExpiredSignatureError:
-            raise HTTPException(
-                status_code=401, detail='Refresh token expired')
+            raise HTTPException(status_code=401, detail="Refresh token expired")
         except jwt.InvalidTokenError:
-            raise HTTPException(
-                status_code=401, detail='Invalid refresh token')
+            raise HTTPException(status_code=401, detail="Invalid refresh token")