Upload 3 files

core/config.py

@@ -0,0 +1,25 @@
+import os
+from pathlib import Path
+from dotenv import load_dotenv
+from urllib.parse import quote_plus
+from pydantic_settings.base import BaseSettingsModel
+
+env_path = Path('.') / '.env'
+load_dotenv(dotenv_path=env_path)
+
+
+class Settings(BaseSettingsModel):
+    # DATABASE_URL: str
+    # DATABASE_NAME: str
+    # DATABASE_USER: str
+    # DATABASE_PASSWORD: str
+    # DATABASE_HOST: str
+    # DATABASE_PORT: str
+    # DATABASE_URL: str
+    # DATABASE_URL = f"postgresql://{DATABASE_USER}:{quote_plus(DATABASE_PASSWORD)}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"
+    JWT_SECRET_KEY: str = os.getenv("JWT_SECRET")
+    JWT_ALGORITHM: str = os.getenv("JWT_ALGORITHM")
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = os.getenv("ACCESS_TOKEN")
+
+def get_settings():
+    return Settings()

core/database.py

@@ -0,0 +1,25 @@
+# database.py
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+
+SQLALCHEMY_DATABASE_URL = "postgresql://user:password@localhost/dbname"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL,
+    pool_size=5,
+    pool_pre_ping=True,
+    pool_recycle=300,
+    max_overflow=0
+                       
+                       )
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

core/security.py

@@ -0,0 +1,73 @@
+from passlib.context import CryptContext
+from fastapi.security import OAuth2PasswordBearer
+from fastapi import Depends, HTTPException
+from datetime import timedelta, datetime
+from jose import JWTError, jwt
+from core.config import get_settings
+from users.services import get_user_by_email
+from sqlalchemy.orm import Session
+from core.database import get_db
+
+
+settings = get_settings()
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+oauth2scheme = OAuth2PasswordBearer(tokenUrl="auth/token/")
+
+def get_password_hash(password):
+    return pwd_context.hash(password)
+
+def verify_password(plain_password, hashed_password):
+    return pwd_context.verify(plain_password, hashed_password)
+
+async def create_access_token(data:dict, expiry:timedelta):
+    payload = data.copy()
+    expire = datetime.utcnow() + expiry
+    payload.update({"exp": expire})
+    token = jwt.encode(payload,
+                       settings.JWT_SECRET_KEY,
+                       algorithm=settings.JWT_ALGORITHM)
+    
+    return token
+
+async def create_refresh_token(data:dict):
+    payload = data.copy()
+    token = jwt.encode(payload,
+                       settings.JWT_SECRET_KEY,
+                       algorithm=settings.JWT_ALGORITHM)
+    return token
+
+def get_token_payload(token:str):
+    try:
+        payload = jwt.decode(token,
+                             settings.JWT_SECRET_KEY,
+                             algorithms=[settings.JWT_ALGORITHM])
+        return payload
+    except JWTError:
+        return None
+    
+async def get_current_user(token:str = Depends(oauth2scheme), db:Session = Depends(get_db)):
+    try:
+        payload = get_token_payload(token)
+        email = payload.get("sub")
+        if email is None:
+            return HTTPException(status_code=401,
+                                 detail="Invalid Token",
+                                 headers={"WWW-Authenticate": "Bearer"})
+    except JWTError:
+        return HTTPException(status_code=401,
+                             detail="Invalid Token",
+                             headers={"WWW-Authenticate": "Bearer"}
+                             )
+    
+    user = get_user_by_email(email, db=db)
+    if user is None:
+        return HTTPException(status_code=401,
+                             detail="Invalid Token",
+                             headers={"WWW-Authenticate": "Bearer"}
+                             )
+    return user
+
+
+
+