tutorial101 is the one place for high quality web development, Web Design and software development tutorials and Resources programming. Learn cutting edge techniques in web development, design and software development, download source components and participate in the community.
#fastapi-jwt/services.py
import fastapi as _fastapi
import fastapi.security as _security
import jwt as _jwt #pip install python_jwt https://pypi.org/project/python-jwt/
import datetime as _dt
import sqlalchemy.orm as _orm
import passlib.hash as _hash
import database as _database, models as _models, schemas as _schemas
oauth2schema = _security.OAuth2PasswordBearer(tokenUrl="/api/token")
JWT_SECRET = "cairocodersednalan"
def create_database():
return _database.Base.metadata.create_all(bind=_database.engine)
def get_db():
db = _database.SessionLocal()
try:
yield db
finally:
db.close()
async def get_user_by_email(email: str, db: _orm.Session):
return db.query(_models.User).filter(_models.User.email == email).first()
async def create_user(user: _schemas.UserCreate, db: _orm.Session):
user_obj = _models.User(
email=user.email, hashed_password=_hash.bcrypt.hash(user.hashed_password)
)
db.add(user_obj)
db.commit()
db.refresh(user_obj)
return user_obj
async def authenticate_user(email: str, password: str, db: _orm.Session):
user = await get_user_by_email(db=db, email=email)
if not user:
return False
if not user.verify_password(password):
return False
return user
async def create_token(user: _models.User):
user_obj = _schemas.User.from_orm(user)
token = _jwt.encode(user_obj.dict(), JWT_SECRET)
return dict(access_token=token, token_type="bearer")
async def get_current_user(
db: _orm.Session = _fastapi.Depends(get_db),
token: str = _fastapi.Depends(oauth2schema),
):
try:
payload = _jwt.decode(token, JWT_SECRET, algorithms=["HS256"])
user = db.query(_models.User).get(payload["id"])
except:
raise _fastapi.HTTPException(
status_code=401, detail="Invalid Email or Password"
)
return _schemas.User.from_orm(user)
fastapi-jwt/database.py
#fastapi-jwt/database.py
import sqlalchemy as _sql
import sqlalchemy.ext.declarative as _declarative
import sqlalchemy.orm as _orm
DATABASE_URL = "sqlite:///./database.db"
engine = _sql.create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = _orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = _declarative.declarative_base()
fastapi-jwt/models.py
#fastapi-jwt/models.py
import datetime as _dt
import sqlalchemy as _sql
import sqlalchemy.orm as _orm
import passlib.hash as _hash
import database as _database
class User(_database.Base):
__tablename__ = "users"
id = _sql.Column(_sql.Integer, primary_key=True, index=True)
email = _sql.Column(_sql.String, unique=True, index=True)
hashed_password = _sql.Column(_sql.String)
def verify_password(self, password: str):
return _hash.bcrypt.verify(password, self.hashed_password)
fastapi-jwt/schemas.py
#fastapi-jwt/schemas.py
import datetime as _dt
import pydantic as _pydantic
class _UserBase(_pydantic.BaseModel):
email: str
class UserCreate(_UserBase):
hashed_password: str
class Config:
orm_mode = True
class User(_UserBase):
id: int
class Config:
orm_mode = True
Run C:\fastAPI\fastapi-jwt>uvicorn main:app --reload
//react/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_methods=["*"],
allow_headers=["*"],
allow_credentials=True,
allow_origins=["http://localhost:3000"]
)
@app.get('/get_data')
async def get_data():
return {'body': 'Hello World from python fastapi'}
run the FastAPI app
C:\fastAPI\react>uvicorn main:app --reload
with uvicorn using the file_name:app_instance open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
Create Project
C:\react-js>npx create-react-app myreactdev
#loginregister/connection.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from typing import Optional
#dbcon = "postgresql://user:password@postgresserver/db"
dbcon = 'sqlite:///fastapidb.sqlite3'
engine = create_engine(dbcon)
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
Base.metadata.create_all(bind=engine)
DATABASE_URL: Optional[str] = None
SECRET_KEY: Optional[str] = "cairocoders"
def sess_db():
db = SessionFactory()
try:
yield db
finally:
db.close()
Create database Model sqlalchemy
loginregister/models.py
#loginregister/models.py
from sqlalchemy import Column,String,Integer,Boolean,Enum
from schema import Roles
from connection import Base
class UserModel(Base):
__tablename__ = "users"
id=Column(Integer,primary_key=True,index=True)
email=Column(String,unique=True,index=True)
username=Column(String,unique=True,index=True)
password=Column(String,unique=False,index=True)
is_active=Column(Boolean,default=False)
role=Column(Enum(Roles),default="user")
loginregister/repositoryuser.py
#loginregister/repositoryuser.py
from sqlalchemy.orm import Session
from models import UserModel
from typing import Dict,Any
from sqlalchemy.orm import Session
import smtplib
from email.message import EmailMessage
class UserRepository:
def __init__(self,sess:Session):
self.sess: Session=sess
def create_user(self,signup:UserModel) -> bool:
try:
self.sess.add(signup)
self.sess.commit()
except:
return False
return True
def get_user(self):
return self.sess.query(UserModel).all()
def get_user_by_username(self,username:str):
return self.sess.query(UserModel).filter(UserModel.username==username).first()
def update_user(self,id:int,details:Dict[str,Any]) -> bool:
try:
self.sess.query(UserModel).filter(UserModel.id==id).update(details)
self.sess.commit()
except:
return False
return True
def delete_user(self,id:int)-> bool:
try:
self.sess.query(UserModel).filter(UserModel.id==id).delete()
self.sess.commit()
except:
return False
return True
class SendEmailVerify:
def sendVerify(token):
email_address = "cairocoders0711@gmail.com" # type Email
email_password = "cairocoders-ednalan" # If you do not have a gmail apps password, create a new app with using generate password. Check your apps and passwords https://myaccount.google.com/apppasswords
# create email
msg = EmailMessage()
msg['Subject'] = "Email subject"
msg['From'] = email_address
msg['To'] = "clydeymojica0130@gmail.com" # type Email
msg.set_content(
f"""\
verify account
http://localhost:8080/user/verify/{token}
""",
)
# send email
with smtplib.SMTP_SSL('smtp.gmail.com', 465) as smtp:
smtp.login(email_address, email_password)
smtp.send_message(msg)
loginregister/schema.py
#loginregister/schema.py
from datetime import date
from pydantic import BaseModel, EmailStr #pip install pydantic[email]
from enum import Enum
from fastapi import Form
class UserSchema(BaseModel):
email: EmailStr
username: str
password: str
class Config:
orm_mode = True
class Roles(Enum):
user = "user"
admin = "admin"
loginregister/scurity.py
#loginregister/scurity.py
from passlib.context import CryptContext #pip install passlib
from fastapi.security import OAuth2PasswordBearer
from jose import jwt #https://pypi.org/project/python-jose/ = pip install python-jose
from fastapi import Depends,Request
from fastapi import HTTPException
from models import UserModel
JWT_SECRET="cairocoders$§%§$Ednalan"
ALGORITHM="HS256"
ACCESS_TOKEN_EXPIRE_MINUTES=3000
pwd_context=CryptContext(schemes=["bcrypt"],deprecated="auto") #pip install bcrypt
# save token to oauth2_scheme
oauth2_scheme=OAuth2PasswordBearer(tokenUrl="user/signin")
COOKIE_NAME="Authorization"
# create Token
def create_access_token(user):
try:
payload={
"username":user.username,
"email":user.email,
"role":user.role.value,
"active":user.is_active,
}
return jwt.encode(payload,key=JWT_SECRET,algorithm=ALGORITHM)
except Exception as ex:
print(str(ex))
raise ex
# create verify Token
def verify_token(token):
try:
payload=jwt.decode(token,key=JWT_SECRET)
return payload
except Exception as ex:
print(str(ex))
raise ex
# password hash
def get_password_hash(password):
return pwd_context.hash(password)
# password verify
def verify_password(plain_password,hashed_password):
return pwd_context.verify(plain_password,hashed_password)
def get_current_user_from_token(token:str=Depends(oauth2_scheme)):
user= verify_token(token)
return user
def get_current_user_from_cookie(request:Request) -> UserModel:
token=request.cookies.get(COOKIE_NAME)
if token:
user = verify_token(token)
return user
//loginregister/templates/verify.html
{% extends "base.html" %}
{% block title %}Contact Page{% endblock %}
{% block content %}
<div class="wrapper">
<div class="contacts">
<h3> verify your account</h3>
</div>
<div class="form">
<h3>Sign in to your email to verify your account</h3>
<form method="post">
<div>
<a class="signin__link" href="/user/signin">back to Login page</a>
</div>
</form>
</div>
</div>
{% endblock %}
run the FastAPI app
C:\fastAPI\loginregister>uvicorn main:app --reload
with uvicorn using the file_name:app_instance open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
C:\fastAPI\sendmail>uvicorn main:app --reload
with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
C:\fastAPI\simplewebsite>uvicorn main:app --reload
with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
#registration/main.py
from fastapi import FastAPI, Request, Depends, Form, status
from fastapi.templating import Jinja2Templates
import models
from database import engine, sessionlocal
from sqlalchemy.orm import Session
from fastapi import responses
from sqlalchemy.exc import IntegrityError
from fastapi.responses import RedirectResponse
from forms import UserCreateForm
models.Base.metadata.create_all(bind=engine)
templates = Jinja2Templates(directory="templates")
app = FastAPI()
def get_db():
db = sessionlocal()
try:
yield db
finally:
db.close()
@app.get("/")
async def home(request: Request, db: Session = Depends(get_db)):
return templates.TemplateResponse("index.html", {"request": request})
@app.post("/register/")
async def register(request: Request, username: str = Form(...), email: str = Form(...), password: str = Form(...), db: Session = Depends(get_db)):
form = UserCreateForm(request)
await form.load_data()
print("submited")
if await form.is_valid():
try:
print(username)
print(email)
print(password)
total_row = db.query(models.User).filter(models.User.email == email).first()
print(total_row)
if total_row == None:
print("Save")
users = models.User(username=username, email=email, password=password)
db.add(users)
db.commit()
return responses.RedirectResponse(
"/", status_code=status.HTTP_302_FOUND
)
else:
print("taken email")
errors = ["The email has already been taken"]
except IntegrityError:
return {"msg":"Error"}
else:
print("Error Form")
errors = form.errors
return templates.TemplateResponse("index.html", {"request": request, "errors": errors})
registration/database.py
#registration/database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
DB_URL = 'sqlite:///fastapidb.sqlite3'
engine = create_engine(DB_URL, connect_args={'check_same_thread': False})
sessionlocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
registration/models.py
#registration/models.py
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
username = Column(String(150))
email = Column(String(150))
password = Column(String(150))
def __repr__(self):
return '<User %r>' % (self.id)
registration/forms.py
#registration/forms.py
from typing import List
from typing import Optional
from fastapi import Request
class UserCreateForm:
def __init__(self, request: Request):
self.request: Request = request
self.errors: List = []
self.username: Optional[str] = None
self.email: Optional[str] = None
self.password: Optional[str] = None
async def load_data(self):
form = await self.request.form()
self.username = form.get("username")
self.email = form.get("email")
self.password = form.get("password")
async def is_valid(self):
if not self.username or not len(self.username) > 3:
self.errors.append("Username should be > 3 chars")
if not self.email or not (self.email.__contains__("@")):
self.errors.append("Email is required")
if not self.password or not len(self.password) >= 4:
self.errors.append("Password must be > 4 chars")
if not self.errors:
return True
return False
C:\fastAPI\registration>uvicorn main:app --reload
with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
C:\fastAPI\multipleupload>uvicorn main:app --reload
with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
//crud/main.py
FastAPI Simple CRUD - SqlAlchemy
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
return "todo"
@app.post("/todo")
def create_todo():
return "create todo item"
@app.get("/todo/{id}")
def read_todo(id: int):
return "read todo item with id {id}"
@app.put("/todo/{id}")
def update_todo(id: int):
return "update todo item with id {id}"
@app.delete("/todo/{id}")
def delete_todo(id: int):
return "delete todo item with id {id}"
@app.get("/todo")
def read_todo_list():
return "read todo list"
crud/main.py
//crud/main.py
from typing import List
from fastapi import FastAPI, status, HTTPException, Depends
from database import Base, engine, SessionLocal
from sqlalchemy.orm import Session
import models
import schemas
Base.metadata.create_all(engine) # Create the database
# Initialize app
app = FastAPI()
# Helper function to get database session
def get_session():
session = SessionLocal()
try:
yield session
finally:
session.close()
@app.get("/")
def root():
return "todo"
@app.post("/todo", response_model=schemas.ToDo, status_code=status.HTTP_201_CREATED)
def create_todo(todo: schemas.ToDoCreate, session: Session = Depends(get_session)):
tododb = models.ToDo(task = todo.task)
session.add(tododb)
session.commit()
session.refresh(tododb)
return tododb
@app.get("/todo/{id}", response_model=schemas.ToDo)
def read_todo(id: int, session: Session = Depends(get_session)):
todo = session.query(models.ToDo).get(id) # get item with the given id
# check if id exists. If not, return 404 not found response
if not todo:
raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")
return todo
@app.put("/todo/{id}", response_model=schemas.ToDo)
def update_todo(id: int, task: str, session: Session = Depends(get_session)):
todo = session.query(models.ToDo).get(id) # get given id
if todo:
todo.task = task
session.commit()
# check if id exists. If not, return 404 not found response
if not todo:
raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")
return todo
@app.delete("/todo/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_todo(id: int, session: Session = Depends(get_session)):
# get the given id
todo = session.query(models.ToDo).get(id)
# if todo item with given id exists, delete it from the database. Otherwise raise 404 error
if todo:
session.delete(todo)
session.commit()
else:
raise HTTPException(status_code=404, detail=f"todo item with id {id} not found")
return None
@app.get("/todo", response_model = List[schemas.ToDo])
def read_todo_list(session: Session = Depends(get_session)):
todo_list = session.query(models.ToDo).all() # get all todo items
return todo_list
//fastAPI\crud_todo\database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# Create a sqlite engine instance
engine = create_engine("sqlite:///fastapidb.db")
# Create a DeclarativeMeta instance
Base = declarative_base()
# Create SessionLocal class from sessionmaker factory
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)
fastAPI\crud_todo\models.py
//fastAPI\crud_todo\models.py
from sqlalchemy import Column, Integer, String
from database import Base
# Define ToDo class from Base
class ToDo(Base):
__tablename__ = 'todos'
id = Column(Integer, primary_key=True)
task = Column(String(256))
fastAPI\crud_todo\schemas.py
//fastAPI\crud_todo\schemas.py
from pydantic import BaseModel
# Create ToDo Schema (Pydantic Model)
class ToDoCreate(BaseModel):
task: str
# Complete ToDo Schema (Pydantic Model)
class ToDo(BaseModel):
id: int
task: str
class Config:
orm_mode = True
run the FastAPI app
C:\fastAPI\crud_todo>uvicorn main:app --reload
with uvicorn using the file_name:app_instance
open the link on the browser http://127.0.0.1:8000/
http://127.0.0.1:8000/docs
//myapp/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./db.sqlite"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
myapp/models.py
//myapp/models.py
from sqlalchemy import Boolean, Column, Integer, String
from database import Base
class Todo(Base):
__tablename__ = "todos"
id = Column(Integer, primary_key=True, index=True)
title = Column(String)
complete = Column(Boolean, default=False)
