4.2. Database Auth

4.2.1. Install

$ pip install 'passlib[bcrypt]'

4.2.2. Example

import uvicorn
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, Session, declarative_base
from passlib.context import CryptContext
from fastapi import FastAPI, Depends, HTTPException, status
app = FastAPI()


SQLALCHEMY_DATABASE_URL = 'sqlite:///:memory:'

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False})
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)
Base = declarative_base()


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


class UserModel(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String)
    email = Column(String)
    password = Column(String)

Base.metadata.create_all(engine)


class UserRequestSchema(BaseModel):
    username: str
    email: str
    password: str


class UserResponseSchema(BaseModel):
    username: str
    email: str

    class Config:
        from_attributes = True


class Password:
    context = CryptContext(schemes=['bcrypt'], deprecated='auto')

    @classmethod
    def encrypt(cls, plaintext_password):
        return cls.context.hash(plaintext_password)

    @classmethod
    def verify(cls, plaintext_password, hashed_password):
        return cls.context.verify(plaintext_password, hashed_password)


@app.post('/user', response_model=UserResponseSchema, tags=['User'])
def create_user(request: UserRequestSchema, db: Session = Depends(get_db)):
    new_user = UserModel(username=request.username,
                         password=Password.encrypt(request.password),
                         email=request.email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


@app.get('/user/{id}', response_model=UserResponseSchema, tags=['User'])
def get_user(id: int, db: Session = Depends(get_db)):
    if user := db.query(UserModel).filter(UserModel.id == id).first():
        return user
    else:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='User does not exist')


if __name__ == '__main__':
    uvicorn.run('test:app', host='127.0.0.1', port=8000, reload=True)