rewrite application with async operetion

Author: darixsamani

app.py

@@ -1,24 +1,30 @@
 from routes.posts import  PostRouter
 from routes.users import  UserRouter
-from fastapi import FastAPI, Depends
-from auth.jwt_bearer import JWTBearer
-from fastapi import Request
-from middleware.middleware import CollectMiddleware
+from fastapi import FastAPI
+from database.database import engine
 
 
-app = FastAPI()
+app = FastAPI(description="Template for building FastAPI applications with PostgreSQL", contact={"email": "samanidarix@gmail.com"})
 
 
-app.add_middleware(CollectMiddleware, redis_host="localhost", redis_port="6379")
 
 
-token_listener = JWTBearer()
+@app.on_event("startup")
+async def startup():
+    await engine.connect()
+
+@app.on_event("shutdown")
+async def shutdown():
+   await engine.dispose()
+
+
 
 
 @app.get("/", tags=["Root"])
 async def read_root():
     return {"message": "Welcome to this fantastic app."}
 
 
+
 app.include_router(UserRouter, tags=["USERS"], prefix="/users")
-app.include_router(PostRouter, tags=["Posts"], prefix="/posts", dependencies=[Depends(token_listener)],)
+app.include_router(PostRouter, tags=["Posts"], prefix="/posts",)

auth/deps.py

@@ -1,31 +1,31 @@
 from fastapi import Depends, status
-from sqlalchemy.orm import Session
-from database.database import SessionLocal
+from sqlalchemy.ext.asyncio import AsyncSession
+from database.database import AsyncSessionLocal
 from auth.jwt_handler import decode_jwt
 from fastapi.security import OAuth2PasswordBearer
 from fastapi.exceptions import HTTPException
 from database.dao.dao_users import  DaoUser
 
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
+async def get_db():
+    async with AsyncSessionLocal() as session:
+        yield session
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/users/token")
 
-def get_current_user(db: Session = Depends(get_db),  token :str = Depends(oauth2_scheme)):
+async def get_current_user(db: AsyncSession = Depends(get_db),  token :str = Depends(oauth2_scheme)):
 
     users_exits = decode_jwt(token=token)
 
-    print(f"user exists : {users_exits}")
-
     if not users_exits:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")
 
     daoUser = DaoUser(db=db)
+
+    user = await daoUser.get_user_by_email(email=users_exits["email"])
+
+    if not user:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials")
 
-    return daoUser.get_user_by_email(email=users_exits["email"])
+    return user
 
 

database/dao/dao_posts.py

@@ -1,41 +1,45 @@
-from sqlalchemy.orm import Session
+from sqlalchemy.future import select
 from models.posts import Post
 from schemas.posts import PostCreate
-from database.database import SessionLocal
+from sqlalchemy.ext.asyncio import AsyncSession
 
 
 class DaoPost():
 
-    db : Session
+    db : AsyncSession
 
 
-    def __init__(self, db: Session):
+    def __init__(self, db: AsyncSession):
         self.db = db
 
 
-    def get_post_user(self, user_id: int):
-        posts_user = self.db.query(Post).filter(Post.user_id==user_id).all()
-        return posts_user
-
-    def get_post_by_id(self, post_id: int):
-        post = self.db.query(Post).filter(Post.id==post_id).first()
+    async def get_post_user(self, user_id: int):
+        result = await self.db.execute(select(Post).where(Post.user_id == user_id))
+        posts = result.scalars()
+        return posts
+
+    async def get_post_by_id(self, post_id: int):
+        result = await self.db.execute(select(Post).where(Post.id == post_id))
+        post = result.scalar_one_or_none()
         return post
 
-    def get_post_by_user_id(self, user_id:int):
-        post = self.db.query(Post).filter(Post.user_id==user_id).first()
-        return post
+    async def get_post_by_user_id(self, user_id:int):
+
+        result = await self.db.execute(select(Post).filter(Post.user_id == user_id))
+        posts = result.scalars()
+        return posts
 
 
 
-    def create_post(self, post_create:PostCreate, user_id: int):
+    async def create_post(self, post_create:PostCreate, user_id: int):
         post = Post(title=post_create.title, content=post_create.content, user_id=user_id)
         self.db.add(post)
-        self.db.commit()
-        self.db.refresh(post)
+        await self.db.commit()
+        await self.db.refresh(post)
 
         return post
 
-    def delete_post(self, post: Post):
+    async def delete_post(self, post: Post):
 
-        self.db.delete(post)
-        self.db.commit()
+        await self.db.delete(post)
+        await self.db.commit()

database/dao/dao_users.py

@@ -1,10 +1,11 @@
-from sqlalchemy.orm import Session
+from sqlalchemy.future import select
+from sqlalchemy.ext.asyncio import AsyncSession
 from models.users import User
 from pydantic import EmailStr
 from passlib.context import CryptContext
 from schemas.users import UserCreate
 from models.users import User
-from database.database import SessionLocal
+import asyncio
 
 hash_helper = CryptContext(schemes=["bcrypt"])
 
@@ -13,40 +14,45 @@
 
 class DaoUser():
 
-    db : Session
+    db : AsyncSession
 
-    def __init__(self, db: Session) -> None:
+    def __init__(self, db: AsyncSession) -> None:
         self.db = db
 
-    def get_users(self):
+    async def get_users(self):
         users = self.db.query(User).all()
         return users
 
 
-    def get_user_by_email(self, email: str):
-        user = self.db.query(User).filter(User.email==email).first()
+    async def get_user_by_email(self, email: str):
+
+        result = await self.db.execute(select(User).where(User.email == email))
+        user = result.scalar_one_or_none()
         return user
 
-    def get_user_by_id(self, user_id: int):
-        user = self.db.query(User).filter(User.id==user_id).first()
+    async def get_user_by_id(self, user_id: int):
+        result = await self.db.execute(select(User).where(User.id == user_id))
+        user = result.scalar_one_or_none()
         return user
 
 
-    def create_new_user(self, user: UserCreate):
+    async def create_new_user(self, user: UserCreate):
         user_db = User(email=user.email, fullname=user.fullname, password = hash_helper.encrypt(user.password))
         self.db.add(user_db)
-        self.db.commit()
-        self.db.refresh(user_db)
+        await self.db.commit()
+        await self.db.refresh(user_db)
         return user_db
 
-    def get_users(self, skip: int = 0, limit: int = 100):
-        return self.db.query(User).offset(skip).limit(limit).all()
+    async def get_users(self, skip: int = 0, limit: int = 100):
+        result = await self.db.execute(select(User).offset(skip).limit(limit))
+        users = result.scalars()
+        return users
 
 
-    def delete_user(self, user):
+    async def delete_user(self, user):
 
-        self.db.delete(user)
-        self.db.commit()
+        await self.db.delete(user)
+        await self.db.commit()
 
 
 

database/database.py

@@ -1,7 +1,8 @@
 import dotenv
 from sqlalchemy import create_engine
 from sqlalchemy.ext.declarative import declarative_base
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
+from typing import Callable
 from config.config import settings
 import os
 
@@ -14,14 +15,17 @@ def get_url():
     server = os.getenv("POSTGRES_SERVER", "localhost")
     db = os.getenv("POSTGRES_DB", "darix")
     port = os.getenv("POSTGRES_PORT", 5432)
-    return f"postgresql://{user}:{password}@{server}:{port}/{db}"
+    return f"postgresql+asyncpg://{user}:{password}@{server}:{port}/{db}"
 
 
 SQLALCHEMY_DATABASE_URL = get_url()
 
 print(f"SQLALCHEMY_DATABASE_URL : {SQLALCHEMY_DATABASE_URL}")
-engine = create_engine(SQLALCHEMY_DATABASE_URL)
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+engine = create_async_engine(SQLALCHEMY_DATABASE_URL)
+
+
+AsyncSessionLocal: Callable[[], AsyncSession] = async_sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)
 
 Base = declarative_base()
 

routes/posts.py

@@ -1,38 +1,39 @@
-from fastapi import APIRouter, Depends
-from sqlalchemy.orm import Session
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.ext.asyncio import AsyncSession
 from auth.deps import get_db, get_current_user
 from schemas.posts import PostCreate
 from models.users import User
 from fastapi import HTTPException
 from database.dao.dao_posts import DaoPost
+from database.dao.dao_users import DaoUser
 from schemas.posts import PostResponseModel
 
 PostRouter = APIRouter()
 
 
 @PostRouter.post("", response_model=PostResponseModel)
-def create_new_post(post_create: PostCreate, db: Session = Depends(get_db), user_create: User = Depends(get_current_user)):
-    print(f"user : {user_create}")
+async def create_new_post(post_create: PostCreate, db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
+
     daoPost = DaoPost(db=db)
-    post = daoPost.create_post(post_create=post_create, user_id=user_create.id)
+    daoUser = DaoUser(db=db)
+    post = await daoPost.create_post(post_create=post_create, user_id=user.id)
 
     if not post:
-
         raise HTTPException(status_code=403, detail="unable to create post")
 
     return post
 
 
 @PostRouter.delete("/{id_post}")
-def delete_posts(id_post:int , db: Session = Depends(get_db), user_create: User = Depends(get_current_user)):
+async def delete_posts(id_post:int,  db: AsyncSession = Depends(get_db), user: User = Depends(get_current_user)):
     daoPost = DaoPost(db=db)
 
-    post = daoPost.get_post_by_id(post_id=id_post)
+    post_exist = await daoPost.get_post_by_id(post_id=id_post)
 
-    if not post:
+    if not post_exist:
 
-        raise HTTPException(status_code=403, detail="Post don't exists")
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Post don't exists")
 
-    daoPost.delete_post(post=post)
+    await daoPost.delete_post(post=post_exist)
 
-    return {"detail": f"Post with id {post.id} successfully deleted"}
+    return HTTPException(status_code=status.HTTP_200_OK, detail=f"Post with id: {post_exist.id} successfully deleted")