user authentication

database.py

@@ -9,4 +9,11 @@
 
 SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False,)
 
-Base = declarative_base()
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

files/repository/blog.py

@@ -0,0 +1,43 @@
+from sqlalchemy.orm import Session
+import models, schemas
+from fastapi import HTTPException,status
+
+def get_all(db: Session):
+    blogs = db.query(models.Blog).all()
+    return blogs
+
+def create(request: schemas.Blog,db: Session):
+    new_blog = models.Blog(title=request.title, body=request.body,user_id=1)
+    db.add(new_blog)
+    db.commit()
+    db.refresh(new_blog)
+    return new_blog
+
+def destroy(id:int,db: Session):
+    blog = db.query(models.Blog).filter(models.Blog.id == id)
+
+    if not blog.first():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Blog with id {id} not found")
+
+    blog.delete(synchronize_session=False)
+    db.commit()
+    return 'done'
+
+def update(id:int,request:schemas.Blog, db:Session):
+    blog = db.query(models.Blog).filter(models.Blog.id == id)
+
+    if not blog.first():
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Blog with id {id} not found")
+
+    blog.update(request)
+    db.commit()
+    return 'updated'
+
+def show(id:int,db:Session):
+    blog = db.query(models.Blog).filter(models.Blog.id == id).first()
+    if not blog:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Blog with the id {id} is not available")
+    return blog

files/repository/user.py

@@ -0,0 +1,19 @@
+
+from sqlalchemy.orm import Session
+import models, schemas
+from fastapi import HTTPException,status
+from hashing import Hash
+
+def create(request: schemas.User,db:Session):
+    new_user = models.User(name=request.name,email=request.email,password=Hash.bcrypt(request.password))
+    db.add(new_user)
+    db.commit()
+    db.refresh(new_user)
+    return new_user
+
+def show(id:int,db:Session):
+    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 the id {id} is not available")
+    return user

files/routers/authentication.py

@@ -0,0 +1,19 @@
+from fastapi import APIRouter, Depends, status, HTTPException
+from fastapi.security import OAuth2PasswordRequestForm
+import schemas, database, models, token
+from hashing import Hash
+from sqlalchemy.orm import Session
+router = APIRouter(tags=['Authentication'])
+
+@router.post('/login')
+def login(request:OAuth2PasswordRequestForm = Depends(), db: Session = Depends(database.get_db)):
+    user = db.query(models.User).filter(models.User.email == request.username).first()
+    if not user:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Invalid Credentials")
+    if not Hash.verify(user.password, request.password):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
+                            detail=f"Incorrect password")
+
+    access_token = token.create_access_token(data={"sub": user.email})
+    return {"access_token": access_token, "token_type": "bearer"}

files/routers/blog.py

@@ -0,0 +1,35 @@
+from typing import List
+from fastapi import APIRouter,Depends,status,HTTPException
+import schemas, database, models, oauth2
+from sqlalchemy.orm import Session
+from ..repository import blog
+
+router = APIRouter(
+    prefix="/blog",
+    tags=['Blogs']
+)
+
+get_db = database.get_db
+
+@router.get('/', response_model=List[schemas.ShowBlog])
+def all(db: Session = Depends(get_db),current_user: schemas.User = Depends(oauth2.get_current_user)):
+    return blog.get_all(db)
+
+
+@router.post('/', status_code=status.HTTP_201_CREATED,)
+def create(request: schemas.Blog, db: Session = Depends(get_db),current_user: schemas.User = Depends(oauth2.get_current_user)):
+    return blog.create(request, db)
+
+@router.delete('/{id}', status_code=status.HTTP_204_NO_CONTENT)
+def destroy(id:int, db: Session = Depends(get_db),current_user: schemas.User = Depends(oauth2.get_current_user)):
+    return blog.destroy(id,db)
+
+
+@router.put('/{id}', status_code=status.HTTP_202_ACCEPTED)
+def update(id:int, request: schemas.Blog, db: Session = Depends(get_db),current_user: schemas.User = Depends(oauth2.get_current_user)):
+    return blog.update(id,request, db)
+
+
+@router.get('/{id}', status_code=200, response_model=schemas.ShowBlog)
+def show(id:int, db: Session = Depends(get_db),current_user: schemas.User = Depends(oauth2.get_current_user)):
+    return blog.show(id,db)

files/routers/user.py

@@ -0,0 +1,21 @@
+from fastapi import APIRouter
+import database, schemas, models
+from sqlalchemy.orm import Session
+from fastapi import APIRouter,Depends,status
+from ..repository import user
+
+router = APIRouter(
+    prefix="/user",
+    tags=['Users']
+)
+
+get_db = database.get_db
+
+
+@router.post('/', response_model=schemas.ShowUser)
+def create_user(request: schemas.User,db: Session = Depends(get_db)):
+    return user.create(request,db)
+
+@router.get('/{id}',response_model=schemas.ShowUser)
+def get_user(id:int,db: Session = Depends(get_db)):
+    return user.show(id,db)

hashing.py

@@ -1,8 +1,10 @@
 from passlib.context import CryptContext
 
-pwd_cxt=CryptContext(schemes=['bcrypt'], deprecated='auto')
+pwd_cxt = CryptContext(schemes=["bcrypt"], deprecated="auto")
 
 class Hash():
-    def bcrypt(password : str):
+    def bcrypt(password: str):
         return pwd_cxt.hash(password)
-
+
+    def verify(hashed_password,plain_password):
+        return pwd_cxt.verify(plain_password,hashed_password)

main.py

@@ -1,8 +1,8 @@
-from fastapi import FastAPI, Depends, status, Response, HTTPException
-from typing import List
-import models, schemas, hashing
-from database import engine, SessionLocal
-from sqlalchemy.orm import Session
+from fastapi import FastAPI, status
+import models
+from database import engine
+from files.routers import blog, user, authentication
+
 
 
 app = FastAPI(
@@ -12,82 +12,12 @@
     description="This API is Created by the developer Manan Ahmed Broti. Not for commercial purpose. The Base URL for this API is set to: https://blog-rhhb.onrender.com/",
 )
 models.Base.metadata.create_all(bind=engine)
+app.include_router(authentication.router)
+app.include_router(blog.router)
+app.include_router(user.router)
 
-def get_db():
-    db = SessionLocal()
-    try:
-        yield db
-    finally:
-        db.close()
 
 @app.get("/", status_code=status.HTTP_200_OK)
 async def credits():
     return {"message": "Welcome to the Blog API. Developed By MAnan Ahmed Broti. Website: AhmedManan.com"}
 
-@app.get('/blog', response_model=List[schemas.ShowBlog], tags=['Blogs'])
-def all_posts(db : Session = Depends(get_db)):
-    blogs = db.query(models.Blog).all()
-    return blogs
-
-@app.post('/blog', status_code = 201, tags=['Blogs'])
-def create_post(request : schemas.Blog, db : Session = Depends(get_db)):
-    new_blog=models.Blog(title=request.title,body=request.body)
-    db.add(new_blog)
-    db.commit()
-    db.refresh(new_blog)
-    return new_blog
-
-@app.get('/blog/{blog_id}', status_code=status.HTTP_200_OK, response_model=schemas.ShowBlog, tags=['Blogs'])
-def get_post(blog_id, response : Response, db : Session = Depends(get_db)):
-    blog = db.query(models.Blog).filter(models.Blog.id== blog_id).first()
-    if not blog:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f'Post not available')
-        # response.status_code = status.HTTP_404_NOT_FOUND
-        # return {'Details':'Blog post not found!'}
-    return blog
-
-@app.put('/blog/{blog_id}', status_code=status.HTTP_202_ACCEPTED, tags=['Blogs'])
-def update_post(blog_id, request : schemas.Blog, db : Session = Depends(get_db)):
-    blog = db.query(models.Blog).filter(models.Blog.id== blog_id)
-
-    if not blog.first():
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f'Post with ID={blog_id} not found')
-
-    blog.update(request.title,request.body)
-    db.commit()
-    return 'updated successfully'
-
-
-@app.delete('/blog/{blog_id}', status_code=status.HTTP_204_NO_CONTENT, tags=['Blogs'])
-def delete_post(blog_id, db : Session = Depends(get_db)):
-        blog = db.query(models.Blog).filter(models.Blog.id== blog_id)
-
-        if not blog.first():
-            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f'Post with ID={blog_id} not found')
-
-        blog.delete(synchronize_session=False)
-        db.commit()
-        return {f'Blog post deleted!'}
-
-@app.post('/user', status_code = 201, tags=['User'])
-def create_user(request: schemas.User, db: Session = Depends(get_db)):
-    # hashed_password = pwd_context.hash(request.password)  # Hash the password
-
-    user_data = request.dict()
-    user_data["password"] = hashing.Hash.bcrypt(request.password)
-
-    new_user = models.User(**user_data)
-    db.add(new_user)
-    db.commit()
-    db.refresh(new_user)
-
-    return new_user
-
-@app.get('/user/{user_id}', status_code=status.HTTP_200_OK, response_model=schemas.ShowUser, tags=['User'])
-def get_post(user_id:int, response : Response, db : Session = Depends(get_db)):
-    user = db.query(models.User).filter(models.User.id== user_id).first()
-    if not user:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,detail=f'User not available')
-        # response.status_code = status.HTTP_404_NOT_FOUND
-        # return {'Details':'Blog post not found!'}
-    return user

models.py

@@ -1,24 +1,25 @@
-from sqlalchemy import Column, Integer, String, DateTime, func
+from sqlalchemy import Column, Integer, String, ForeignKey
 from database import Base
+from sqlalchemy.orm import relationship
+
 
 class Blog(Base):
     __tablename__ = 'blogs'
+
     id = Column(Integer, primary_key=True, index=True)
     title = Column(String)
     body = Column(String)
-    author = Column(String)
-    tags = Column(String)
-    created_at = Column(DateTime, default=func.now())
-    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+    user_id = Column(Integer, ForeignKey('users.id'))
+
+    creator = relationship("User", back_populates="blogs")
 
 
 class User(Base):
     __tablename__ = 'users'
+
     id = Column(Integer, primary_key=True, index=True)
     name = Column(String)
-    email = Column(String)  # Fixed typo here
+    email = Column(String)
     password = Column(String)
-    user_type = Column(String)  # Renamed "type" to "user_type"
-    status = Column(String)
-    created_at = Column(DateTime, default=func.now())
-    updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    blogs = relationship('Blog', back_populates="creator")

my_token.py

@@ -0,0 +1,24 @@
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+import schemas
+
+SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+def create_access_token(data: dict):
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+def verify_token(token:str,credentials_exception):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        email: str = payload.get("sub")
+        if email is None:
+            raise credentials_exception
+        token_data = schemas.TokenData(email=email)
+    except JWTError:
+        raise credentials_exception

oauth2.py

@@ -0,0 +1,14 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+import token
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
+
+def get_current_user(data: str = Depends(oauth2_scheme)):
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+
+    return token.verify_token(data, credentials_exception)

requirements.txt

@@ -5,4 +5,5 @@ mysql-connector-python
 passlib
 bcrypt
 python-jose
-python-multipart
+python-multipart
+cryptography