Add authentication endpoints and JWT handling for user login

app/config.py

@@ -10,6 +10,9 @@ class Settings(BaseSettings):
     DB_PASS: str = ""
     DB_NAME: str = ""
     DB_ECHO: bool = False
+    SECRET_KEY: str = "change-me-to-a-random-secret-key"
+    ALGORITHM: str = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES: int = 1440
 
     @property
     def DATABASE_URL(self) -> str:

app/routers/auth.py

@@ -1,15 +1,35 @@
 from fastapi import APIRouter, Depends, HTTPException, status
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, selectinload
 
 from app.database import get_db
 from app.models.user import User
 from app.models.user_profile import UserProfile
-from app.schemas.user import UserCreate, UserResponse
+from app.schemas.user import LoginRequest, LoginResponse, UserCreate, UserResponse
-from app.security import hash_password
+from app.security import create_access_token, hash_password, verify_password
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
 
+@router.post("/login", response_model=LoginResponse)
+def login(data: LoginRequest, db: Session = Depends(get_db)):
+    user = db.query(User).options(selectinload(User.profile)).filter(
+        (User.email == data.login) | (User.username == data.login),
+        User.deleted_at.is_(None),
+    ).first()
+
+    if not user or not verify_password(data.password, user.password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail={"message": "Kredensial autentikasi tidak ada atau tidak valid."},
+        )
+
+    access_token = create_access_token({"sub": user.id})
+    return LoginResponse(
+        access_token=access_token,
+        user=user,
+    )
+
+
 @router.post("/register", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
 def register(data: UserCreate, db: Session = Depends(get_db)):
     existing = db.query(User).filter(
@@ -18,7 +38,7 @@ def register(data: UserCreate, db: Session = Depends(get_db)):
     if existing:
         raise HTTPException(
             status_code=status.HTTP_409_CONFLICT,
-            detail={"message": "Email or username already registered"},
+            detail={"message": "Email atau username sudah terdaftar."},
         )
 
     user = User(

app/schemas/__init__.py

@@ -1,4 +1,4 @@
-from app.schemas.user import UserCreate, UserResponse, UserDetailResponse, UserProfileResponse, UserUpdate
+from app.schemas.user import UserCreate, UserResponse, UserDetailResponse, UserProfileResponse, UserUpdate, LoginRequest, LoginResponse
 from app.schemas.business_type import BusinessTypeCreate, BusinessTypeResponse, BusinessTypeUpdate
 from app.schemas.plan import PlanCreate, PlanResponse, PlanUpdate
 from app.schemas.tenant import TenantCreate, TenantResponse, TenantUpdate
@@ -10,6 +10,8 @@ __all__ = [
     "UserDetailResponse",
     "UserProfileResponse",
     "UserUpdate",
+    "LoginRequest",
+    "LoginResponse",
     "BusinessTypeCreate",
     "BusinessTypeResponse",
     "BusinessTypeUpdate",

app/schemas/user.py

@@ -45,3 +45,14 @@ class UserProfileResponse(BaseModel):
 
 class UserDetailResponse(UserResponse):
     profile: UserProfileResponse | None = None
+
+
+class LoginRequest(BaseModel):
+    login: str
+    password: str
+
+
+class LoginResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    user: UserDetailResponse

app/security.py

@@ -1,4 +1,9 @@
+from datetime import datetime, timedelta, timezone
+
 import bcrypt
+from jose import JWTError, jwt
+
+from app.config import settings
 
 
 def hash_password(password: str) -> str:
@@ -7,3 +12,17 @@ def hash_password(password: str) -> str:
 
 def verify_password(plain: str, hashed: str) -> bool:
     return bcrypt.checkpw(plain.encode(), hashed.encode())
+
+
+def create_access_token(data: dict) -> str:
+    to_encode = data.copy()
+    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+
+
+def decode_access_token(token: str) -> dict | None:
+    try:
+        return jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
+    except JWTError:
+        return None

requirements.txt

@@ -3,6 +3,7 @@ annotated-types==0.7.0
 anyio==4.14.1
 certifi==2026.6.17
 click==8.4.2
+cryptography==44.0.0
 detect-installer==0.1.0
 dnspython==2.8.0
 email-validator==2.3.0
@@ -21,6 +22,7 @@ markdown-it-py==4.2.0
 MarkupSafe==3.0.3
 mdurl==0.1.2
 pydantic==2.13.4
+python-jose==3.3.0
 pydantic-extra-types==2.11.1
 pydantic-settings==2.14.2
 pydantic_core==2.46.4