Sesi 3: Models & Auth System

from datetime import datetime from sqlalchemy import Column, Integer, DateTime from db import Base class TimestampMixin: """Mixin untuk kolom created_at yang sama di semua tabel""" created_at = Column(DateTime, default=datetime.utcnow, nullable=False)

from sqlalchemy import Column, Integer, String, Boolean, ForeignKey, Table from sqlalchemy.orm import relationship from app.models.base import TimestampMixin, Base # Tabel Junction untuk Many-to-Many (Users <-> Roles) user_role = Table( 'user_role', Base.metadata, Column('user_id', Integer, ForeignKey('users.id', ondelete="CASCADE"), primary_key=True), Column('role_id', Integer, ForeignKey('roles.id', ondelete="CASCADE"), primary_key=True) ) class Branch(Base): __tablename__ = "branches" id = Column(Integer, primary_key=True, index=True) code = Column(String(10), unique=True, nullable=False, index=True) name = Column(String(100), nullable=False) city = Column(String(100), nullable=False) is_active = Column(Boolean, default=True) class Role(Base): __tablename__ = "roles" id = Column(Integer, primary_key=True, index=True) name = Column(String(50), unique=True, nullable=False) description = Column(String(255)) class User(Base, TimestampMixin): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) username = Column(String(100), unique=True, nullable=False, index=True) password_hash = Column(String(255), nullable=False) full_name = Column(String(150), nullable=False) email = Column(String(150)) phone = Column(String(30)) branch_id = Column(Integer, ForeignKey("branches.id")) is_active = Column(Boolean, default=True) # Relationships branch = relationship("Branch", backref="users") roles = relationship("Role", secondary=user_role, back_populates="users") # Setup reverse relationship for Role -> User (Opsional tapi bagus untuk lazy loading) Role.users = relationship("User", secondary=user_role, back_populates="roles")

from pydantic import BaseModel, EmailStr from typing import Optional, List from datetime import datetime # --- Auth Schemas --- class LoginRequest(BaseModel): username: str password: str class Token(BaseModel): access_token: str token_type: str class TokenData(BaseModel): username: Optional[str] = None # --- User Schemas --- class RoleBase(BaseModel): name: str description: Optional[str] = None class RoleSchema(RoleBase): id: int class Config: from_attributes = True class UserBase(BaseModel): username: str full_name: str email: Optional[EmailStr] = None phone: Optional[str] = None class UserCreate(UserBase): password: str branch_id: Optional[int] = None class UserResponse(UserBase): id: int is_active: bool branch_id: Optional[int] = None created_at: datetime roles: List[RoleSchema] = [] class Config: from_attributes = True # Penting untuk Pydantic V2 + SQLAlchemy

from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer from sqlalchemy.orm import Session from jose import JWTError, jwt from db import get_db from security import verify_password from config import get_settings from app.models.user import User # OAuth2 Scheme: Client mengirim header "Authorization: Bearer " oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/v1/auth/login") settings = get_settings() def get_user_by_username(db: Session, username: str) -> User: return db.query(User).filter(User.username == username).first() def authenticate_user(db: Session, username: str, password: str): user = get_user_by_username(db, username) if not user: return False if not verify_password(password, user.password_hash): return False return user async def get_current_user( token: str = Depends(oauth2_scheme), db: Session = Depends(get_db) ): 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]) username: str = payload.get("sub") if username is None: raise credentials_exception except JWTError: raise credentials_exception user = get_user_by_username(db, username=username) if user is None: raise credentials_exception return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if not current_user.is_active: raise HTTPException(status_code=400, detail="Inactive user") return current_user

from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy.orm import Session from datetime import timedelta from db import get_db from security import create_access_token from config import get_settings from app.schemas.auth import LoginRequest, Token, UserResponse from app.api.deps import authenticate_user, get_current_active_user router = APIRouter() settings = get_settings() @router.post("/login", response_model=Token) def login(user_credentials: LoginRequest, db: Session = Depends(get_db)): """ Endpoint Login. Mengembalikan JWT Access Token. """ user = authenticate_user(db, user_credentials.username, user_credentials.password) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}, ) access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) access_token = create_access_token( data={"sub": user.username}, expires_delta=access_token_expires ) return {"access_token": access_token, "token_type": "bearer"} @router.get("/me", response_model=UserResponse) def read_users_me(current_user: UserResponse = Depends(get_current_active_user)): """ Endpoint Get Profile User saat ini. Membutuhkan Header Authorization: Bearer """ return current_user

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from db import engine, get_db import config # Import Router Auth from app.api.auth import router as auth_router app = FastAPI(title=config.get_settings().APP_NAME, version="1.0.0") # CORS app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Include Routes app.include_router(auth_router, prefix="/api/v1/auth", tags=["Authentication"]) @app.on_event("startup") def on_startup(): try: with engine.connect() as conn: print("✅ DB Connected!") except Exception as e: print(f"❌ DB Error: {e}") @app.get("/") def root(): return {"status": "BPRS API Running", "session": "3"}

from db import SessionLocal from security import get_password_hash from app.models.user import User def reset_admin_password(): db = SessionLocal() try: # Cari user admin admin = db.query(User).filter(User.username == "admin").first() if admin: new_hash = get_password_hash("Admin@Baktimakmur#2026") admin.password_hash = new_hash db.commit() print(f"✅ Password untuk user '{admin.username}' berhasil direset!") else: print("❌ User admin tidak ditemukan.") finally: db.close() if __name__ == "__main__": reset_admin_password()

python init_passwords.py

python main.py

http://IP-SERVER:8000/docs