Users table creation inFastAPI

Creating user table code and recreating database:

models.py file content:

# This will help sqlalchemy to understand the database table and work on it.
from database import Base
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key = True, index = True)
    email = Column(String, unique = True)
    username = Column(String, unique = True)
    first_name = Column(String)
    last_name = Column(String)
    hashed_password = Column(String) # This will have encypted
password which will never get decrypted.
    is_active = Column(Boolean, default = True) # Default value is True,
meaning the user is active by default.
    role = Column(String)

class Todos(Base):
    __tablename__ = 'todos'

    id = Column(Integer, primary_key = True, index = True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    complete = Column(Boolean, default = False)
    owner_id = Column(Integer, ForeignKey('users.id'))  # Foreign key to
link to the Users table

Running uvicorn main:app --reload command to recreate the database after deleting it.

Post route creation for Users entry:

UsersValidator.py - pydantic file to validate input data

# This document is created to implement pydantic validators to accept right
values for the Users table.
from pydantic import BaseModel, EmailStr, Field

class UserRequest(BaseModel):
    email: EmailStr
    username: str = Field(..., min_length=3, max_length=50)
    first_name: str = Field(..., min_length=2, max_length=50)
    last_name: str = Field(..., min_length=2, max_length=50)
    password: str = Field(..., min_length=6)
    role: str = Field(..., min_length=3, max_length=20)

Updating auth.py file for new route details. Taking plain password as input at the moment.

# This file is to define authenticator and authorization for the Todo application.
from venv import create
from fastapi import APIRouter,status
from UsersValidator import UserRequest
from models import Users

router = APIRouter()

@router.get("/auth", status_code=status.HTTP_200_OK)
async def get_user():
    return {'user': 'authenticated_user'}

@router.post("/auth/createUser", status_code=status.HTTP_200_OK)
async def create_user(create_user_request: UserRequest):
    # create_user_model = create_user_request.model_dump()
but it will not work as table has hashed_password and input as plain password.
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        hashed_password=create_user_request.password,
        role=create_user_request.role,
        is_active=True  # Default value for new users
    )

    return create_user_model

Hashing the password:

Amended the auth.py file:
# This file is to define authenticator and authorization for the Todo application.
from venv import create
from fastapi import APIRouter,status
from UsersValidator import UserRequest
from models import Users

router = APIRouter()

# Method to encrypt the password using bcrypt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

@router.get("/auth", status_code=status.HTTP_200_OK)
async def get_user():
    return {'user': 'authenticated_user'}

@router.post("/auth/createUser", status_code=status.HTTP_200_OK)
async def create_user(create_user_request: UserRequest):
    # create_user_model = create_user_request.model_dump()
but it will not work as table has hashed_password and input as plain password.
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        hashed_password=pwd_context.hash(create_user_request.password),
        role=create_user_request.role,
        is_active=True  # Default value for new users
    )

    return create_user_model

Save user to Database:

# This file is to define authenticator and authorization for the Todo application.
from venv import create
from fastapi import APIRouter, status, Depends, HTTPException
from UsersValidator import UserRequest
from models import Users
from database import SessionLocal
from sqlalchemy.orm import Session
from typing import Annotated

router = APIRouter()

# Method to encrypt the password using bcrypt
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# creating db dependency
def get_db():
    # We need to fetch this SessionLocal before each request made to database.
    db = SessionLocal()
    try:
        # Only the code prio to and including the
        # yield statement will execute before sending the response.
        yield db
    finally:
        # It will be executed after sending the response.
        # It makes fastAPI quicker as we can fetch information
        # from the database and return it to client
        # and then can close the db connection.
        # It makes the connection extreamly safe.
        # As it will open the database connect only at the time of usage.
        db.close()

db_dependency = Annotated[Session, Depends(get_db)]

@router.get("/auth", status_code=status.HTTP_200_OK)
async def get_user():
    return {'user': 'authenticated_user'}

@router.post("/auth/createUser", status_code=status.HTTP_201_CREATED)
async def create_user(db: db_dependency,
                      create_user_request: UserRequest):
    # create_user_model = create_user_request.model_dump() but it
    # will not work as table has hashed_password and input as plain password.
    create_user_model = Users(
        email=create_user_request.email,
        username=create_user_request.username,
        first_name=create_user_request.first_name,
        last_name=create_user_request.last_name,
        hashed_password=pwd_context.hash(create_user_request.password),
        role=create_user_request.role,
        is_active=True  # Default value for new users
    )
    db.add(create_user_model)
    db.commit()

Checking table entry in sqlite:

sqllite3 todos.db

select * from users;

Comments

Popular posts from this blog

Post Request with Pydantic usage (input validator)

Code Scalability and Routing

CRUD Assignment