Skip to content

josue-lubaki/FastAPI-Python-Tutorial

Repository files navigation

Heroku : https://fastapi-tuto.herokuapp.com/
Production : https://sappserver.xyz/
Course : https://www.youtube.com/watch?v=0sOvCWFmrtA&t=23834s&ab_channel=freeCodeCamp.org

  • Create a virtual environment > py -3 -m venv venv
  • Activate bat environment >.\venv\Scripts\activate
  • Install FastApi > pip install fastApi[all]
  • Install All require dependencies > pip install -r requirements.txt
  • Import fastApi into project
from fastapi import FastAPI
app = FastAPI()

@app.get("/")
async def read_root():
    return {"Message": "Hello World"}
  • Run the fastApi module > uvicorn app.main:app --reload
  • Field Types on Python using Pydantic Library
class User(BaseModel):
    id: int
    name = 'John Doe'
    signup_ts: Optional[datetime] = None
    friends: List[int] = []
  • Import Pyscopg library form postgres/python
   import psycopg2
   from psycopg2.extras import RealDictCursor

   # Connect to an existing database
   conn = psycopg2.connect(host='localhost', database='databaseName', user='postgres', password='password', cursor_factory=RealDictCursor, port=5432)
   
   # Open a cursor to perform database operations
   cursor = conn.cursor()

   # Execute a command: this creates a new table
   cursor.execute("CREATE TABLE test (id serial PRIMARY KEY, num integer, data varchar);")
  • Importing database and run it :
while True:
   try:
       conn = psycopg2.connect(
           host='localhost', database='databaseName', user='postgres', password='password', cursor_factory=RealDictCursor, port=5433)
       cursor = conn.cursor()
       print("Database connection was successfull !")
       break
   except Exception as error:
       print("Connecting to database failed")
       print("Error : ", error)
       time.sleep(2)
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# SQLALCHEMY_DATABASE_URL = "postgresql://<username>:<password>@<ip-adress/hostName>:<port>/<databaseName>"
SQLALCHEMY_DATABASE_URL = "postgresql://postgres:password@localhost:5433/fastapi"

engine = create_engine(SQLALCHEMY_DATABASE_URL)

SessionLocator = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
  • Validation email & Expose data you want
from pydantic immport EmailStr

# retrieve exactly the field you want to expose
class Post(PostBase):
    id: int
    created_at: datetime

    class Config:
        orm_mode = True
  • Hashing Password user with JWT
from passlib.context import CryptContext

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

@app.post("/users", status_code=status.HTTP_201_CREATED, response_model=schemas.UserOut)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    # if user exists
    new_user = db.query(models.User).filter(
        models.User.email == user.email).first()

    if new_user:
        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST,
                            detail=f"User with email: {user.email} already exists")

    # hash the password
    hashed_password = pwd_context.hash(user.password)
    user.password = hashed_password

    new_user = models.User(**user.dict())
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

  • Create a router
# in router file (routers/user.py):
from fastapi import APIRouter

router = APIRouter(
    prefix='/users',
    tags=['Users']
)

@router.get('/{id}', response_model=schemas.UserOut)
def get_user(id: int, db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.id == id).first()

    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
                            detail=f"User with id: {id} was not found")

    return user


# in the main.py file:

from .routers import user
app = FastAPI()
app.include_router(user.router)
# in the auth.py file:
from fastapi import APIRouter, Depends, HTTPException, Response, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from .. import database, schemas, models, utils, oauth2

router = APIRouter(
    tags=['Authentication']
)

@router.post('/login')
def login(user_credentials: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
    # change email of user_credentials to user_credentials.username because, i used OAuth2PasswordRequestForm
    user = db.query(models.User).filter(
        models.User.email == user_credentials.username).first()

    if not user:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Incorrect email or password")

    if not utils.verify_password(user_credentials.password, user.password):
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN, detail="Incorrect email or password")

    # create a token for the user, and sharing "id" of user
    access_token = oauth2.create_access_token(data={"user_id": user.id})

    return {"access_token": access_token, "token_type": "bearer"}
# in the aouth2.py file :
from jose import JWTError, jwt
from datetime import datetime, timedelta
from . import schemas
from fastapi import Depends, status, HTTPException
from fastapi.security import OAuth2PasswordBearer

# which route protect by bearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

SECRET_KEY = '09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7'
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

# create_access_method
def create_access_token(data: dict):
    to_encode = data.copy()

    # time expires
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})

    encode_jwt = jwt.encode(to_encode, SECRET_KEY, ALGORITHM)

    return encode_jwt


# method that veify your token if it's valid
def verify_access_token(token: str, credentials_exception):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # get same information of sending since login method
        id = payload = payload.get("user_id")

        if id is None:
            raise credentials_exception
        token_data = schemas.TokenData(id=id)
    except JWTError:
        raise credentials_exception

    return token_data
# in the schemas.py file :
class UserLogin(BaseModel):
    email: str
    password: str


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    id: Optional[str] = None


# in the Utils.py file :

# method that will be used to verify passwords
def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)
  • Alembic import into project
# command :> pip install alembic
# command :> alembic init <directoryName>
# command :> alembic revision -m "message"
# command :> alembic heads
# command :> alembic upgrade head || alembic upgrade +1
# command :> alembic downgrade <revisionCode> || alembic downgrade -1
# in the ".env.py" file:
from app.models import Base
from app.config import settings

config.set_main_option(
    "sqlalchemy.url", f"postgresql+psycopg2://{settings.database_username}:{settings.database_password}@{settings.database_hostname}:{settings.database_port}/{settings.database_name}")

target_metadata = Base.metadata
  • Alembic command DDL
# Example revision
def upgrade():
    op.create_table('users',
                    sa.Column('id', sa.Integer(), nullable=False, primary_key=True),
                    sa.Column('email', sa.String(), nullable=False, unique=True),
                    sa.Column('password', sa.String(), nullable=False),
                    sa.Column('created_at', sa.TIMESTAMP(timezone=True),
                              server_default=text('NOW()'), nullable=False)
                    )
    pass


def downgrade():
    op.drop_table('users')
    pass
  • Autogenerate Alembic
Command :> Alembic revision --autogenerate -m "message"
INFO : Don't need @code< models.Base.metadata.create_all(bind=engine) > in the main.py
  • export all required dependencies on the text file & install all dependencies
Command :> pip freeze > "FileName.txt"
Command :> pip install -r "FileName.txt"
from fastapi.middleware.cors import CORSMiddleware

origins = ["*"]

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)
  • Deploy Heroku
- Heroku Postgres : https://devcenter.heroku.com/articles/heroku-postgresql
Command :> heroku addons:create heroku-postgresql:hobby-dev

Command :> heroku login
Command :> heroku create <AppName>
Command :> heroku git:remote -a <AppName>
Command :> git push heroku main
Command :> heroku logs -t
Command :> heroku ps:restart
  • Never run "revision" command on prod
Command :> heroku run "alembic upgrade head"
  • Ubuntu Command
Command :> sudo apt update && sudo apt upgrade -y
Command :> sudo apt install python3-pip
Command :> sudo pip3 install postgresql postgresql-contrib -y
Command :> psql -U <name1>
Command :> \password <name1>
Command :> cd /etc/postgresql/<version>/main
Command :> systemctl restart postgresql
Command :> adduser <name2>
Command :> usermod -aG sudo <name2>
Command :> set -o allexport; source /home/<name2>/.env; set +o allexport
Command :> gunicorn -w 4 -k uvicorn.workers.UvicornWorker app.main:app --bind 0.0.0.0:8000
Command :> sudo snap install --classic certbot
Command :> sudo certbot --nginx
  • Add Firewall configuration
Command :> sudo ufw allow http
Command :> sudo ufw allow https
Command :> sudo ufw allow ssh
Command :> sudo ufw allow 5432
Command :> sudo ufw enable
Command :> sudo ufw status
Command :> sudo ufw delete http

About

Développement d'un API grâce à FASTAPI

Resources

License

Stars

Watchers

Forks