user authentification

Backend/app/api/auth.py

@@ -0,0 +1,62 @@
+from typing import Any
+from fastapi import APIRouter, Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordRequestForm
+from sqlalchemy.orm import Session
+from app.api import deps
+from app.core import security
+from app.models.user import User
+from app.schemas.user import UserCreate, User as UserSchema
+from app.schemas.token import Token
+
+router = APIRouter()
+
+@router.post("/register", response_model=UserSchema)
+def register(
+    *,
+    db: Session = Depends(deps.get_db),
+    user_in: UserCreate,
+) -> Any:
+    """
+    Create new user.
+    """
+    user = db.query(User).filter(User.email == user_in.email).first()
+    if user:
+        raise HTTPException(
+            status_code=400,
+            detail="The user with this username already exists in the system.",
+        )
+
+    hashed_password = security.get_password_hash(user_in.password)
+    db_user = User(
+        email=user_in.email,
+        hashed_password=hashed_password,
+        is_active=user_in.is_active,
+        is_superuser=user_in.is_superuser,
+    )
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+@router.post("/login", response_model=Token)
+def login_access_token(
+    db: Session = Depends(deps.get_db), form_data: OAuth2PasswordRequestForm = Depends()
+) -> Any:
+    """
+    OAuth2 compatible token login, get an access token for future requests
+    """
+    user = db.query(User).filter(User.email == form_data.username).first()
+    if not user or not security.verify_password(form_data.password, user.hashed_password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Incorrect email or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    elif not user.is_active:
+        raise HTTPException(status_code=400, detail="Inactive user")
+
+    access_token = security.create_access_token(subject=user.email)
+    return {
+        "access_token": access_token,
+        "token_type": "bearer",
+    }

Backend/app/api/deps.py

@@ -0,0 +1,42 @@
+from typing import Generator, Optional
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import jwt, JWTError
+from sqlalchemy.orm import Session
+from app.core.config import settings
+from app.database.session import SessionLocal
+from app.models.user import User
+from app.schemas.token import TokenData
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl=f"{settings.API_V1_STR}/auth/login")
+
+def get_db() -> Generator:
+    try:
+        db = SessionLocal()
+        yield db
+    finally:
+        db.close()
+
+def get_current_user(
+    db: Session = Depends(get_db), token: str = Depends(oauth2_scheme)
+) -> User:
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    try:
+        payload = jwt.decode(
+            token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]
+        )
+        email: str = payload.get("sub")
+        if email is None:
+            raise credentials_exception
+        token_data = TokenData(email=email)
+    except JWTError:
+        raise credentials_exception
+
+    user = db.query(User).filter(User.email == token_data.email).first()
+    if user is None:
+        raise credentials_exception
+    return user

Backend/app/core/config.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings
+from typing import Optional
+
+class Settings(BaseSettings):
+    PROJECT_NAME: str = "JobmateAI Backend"
+    API_V1_STR: str = "/api/v1"
+    SECRET_KEY: str = "YOUR_SECRET_KEY_HERE_PLEASE_CHANGE_IT" # TODO: Change this in production
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8 # 8 days
+    ALGORITHM: str = "HS256"
+
+    SQLALCHEMY_DATABASE_URI: Optional[str] = "sqlite:///./sql_app.db"
+
+    class Config:
+        case_sensitive = True
+
+settings = Settings()

Backend/app/core/security.py

@@ -0,0 +1,23 @@
+from datetime import datetime, timedelta
+from typing import Any, Union, Optional
+from jose import jwt
+from passlib.context import CryptContext
+from app.core.config import settings
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+def create_access_token(subject: Union[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+
+    to_encode = {"exp": expire, "sub": str(subject)}
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    return pwd_context.hash(password)

Backend/app/database/base.py

@@ -0,0 +1,12 @@
+from typing import Any
+from sqlalchemy.ext.declarative import as_declarative, declared_attr
+
+@as_declarative()
+class Base:
+    id: Any
+    __name__: str
+
+    # Generate __tablename__ automatically
+    @declared_attr
+    def __tablename__(cls) -> str:
+        return cls.__name__.lower()

Backend/app/database/session.py

@@ -0,0 +1,8 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.core.config import settings
+
+engine = create_engine(
+    settings.SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False}
+)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Backend/app/main.py

@@ -0,0 +1,16 @@
+from fastapi import FastAPI
+from app.core.config import settings
+from app.api import auth
+from app.database.base import Base
+from app.database.session import engine
+
+# Create database tables
+Base.metadata.create_all(bind=engine)
+
+app = FastAPI(title=settings.PROJECT_NAME)
+
+app.include_router(auth.router, prefix=f"{settings.API_V1_STR}/auth", tags=["auth"])
+
+@app.get("/")
+def read_root():
+    return {"message": "Welcome to JobmateAI Backend"}

Backend/app/models/user.py

@@ -0,0 +1,9 @@
+from sqlalchemy import Boolean, Column, Integer, String
+from app.database.base import Base
+
+class User(Base):
+    id = Column(Integer, primary_key=True, index=True)
+    email = Column(String, unique=True, index=True)
+    hashed_password = Column(String)
+    is_active = Column(Boolean, default=True)
+    is_superuser = Column(Boolean, default=False)

Backend/app/schemas/token.py

@@ -0,0 +1,9 @@
+from typing import Optional
+from pydantic import BaseModel
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TokenData(BaseModel):
+    email: Optional[str] = None

Backend/app/schemas/user.py

@@ -0,0 +1,17 @@
+from typing import Optional
+from pydantic import BaseModel, EmailStr
+
+class UserBase(BaseModel):
+    email: Optional[EmailStr] = None
+    is_active: Optional[bool] = True
+    is_superuser: bool = False
+
+class UserCreate(UserBase):
+    email: EmailStr
+    password: str
+
+class User(UserBase):
+    id: int
+
+    class Config:
+        from_attributes = True

Backend/requirements.txt

@@ -1 +1,12 @@
-
+fastapi
+uvicorn
+sqlalchemy
+alembic
+passlib[bcrypt]
+python-jose[cryptography]
+python-multipart
+pydantic-settings
+httpx
+pytest
+email-validator
+bcrypt==4.0.1

Backend/tests/test_auth.py

@@ -0,0 +1,53 @@
+from fastapi.testclient import TestClient
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+from app.main import app
+from app.database.base import Base
+from app.api.deps import get_db
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base.metadata.create_all(bind=engine)
+
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+app.dependency_overrides[get_db] = override_get_db
+
+client = TestClient(app)
+
+def test_create_user():
+    response = client.post(
+        "/api/v1/auth/register",
+        json={"email": "test@example.com", "password": "password123", "is_active": True},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["email"] == "test@example.com"
+    assert "id" in data
+
+def test_login_user():
+    response = client.post(
+        "/api/v1/auth/login",
+        data={"username": "test@example.com", "password": "password123"},
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert "access_token" in data
+    assert data["token_type"] == "bearer"
+
+def test_login_wrong_password():
+    response = client.post(
+        "/api/v1/auth/login",
+        data={"username": "test@example.com", "password": "wrongpassword"},
+    )
+    assert response.status_code == 401