Sesi 2: Backend Core Setup

# Database Configuration DB_HOST=localhost DB_PORT=3306 DB_USER=ao DB_PASSWORD=ao123 DB_NAME=ao # Security Configuration SECRET_KEY=ubah_ini_dengan_random_string_yang_panjang_untuk_production ALGORITHM=HS256 ACCESS_TOKEN_EXPIRE_MINUTES=300 # App Config APP_NAME=BPRS Baktimakmur API APP_ENV=production

from pydantic_settings import BaseSettings from functools import lru_cache class Settings(BaseSettings): # Database DB_HOST: str = "localhost" DB_PORT: int = 3306 DB_USER: str = "ao" DB_PASSWORD: str = "ao123" DB_NAME: str = "ao" # Security SECRET_KEY: str ALGORITHM: str = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES: int = 300 # App APP_NAME: str = "BPRS API" APP_ENV: str = "development" @property def DATABASE_URL(self): return f"mysql+pymysql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}" class Config: env_file = ".env" @lru_cache() def get_settings(): return Settings()

from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from config import get_settings settings = get_settings() # Koneksi ke MySQL # pool_recycle=3600 mencegah error "MySQL has gone away" pada koneksi lama engine = create_engine( settings.DATABASE_URL, pool_pre_ping=True, pool_recycle=3600, echo=False # Set True jika ingin melihat query SQL di log ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # Dependency untuk mendapatkan sesi DB di setiap endpoint def get_db(): db = SessionLocal() try: yield db finally: db.close()

from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from passlib.context import CryptContext from config import get_settings settings = get_settings() # Password Hashing (Bcrypt) pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt

from fastapi import FastAPI, Depends, HTTPException from fastapi.middleware.cors import CORSMiddleware from sqlalchemy.orm import Session from db import engine, get_db import config # Import Base untuk nantinya membuat tabel (di sesi berikutnya) # from models import Base app = FastAPI(title=config.get_settings().APP_NAME, version="1.0.0") # CORS Middleware agar Frontend (React) bisa akses API ini app.add_middleware( CORSMiddleware, allow_origins=["*"], # Di production, ganti dengan domain frontend: https://ao.baktimakmur.com allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) @app.on_event("startup") def on_startup(): # Cek koneksi database saat startup try: with engine.connect() as conn: print("✅ Successfully connected to MySQL Database!") except Exception as e: print(f"❌ Database connection failed: {e}") @app.get("/") def read_root(): return { "status": "success", "message": "BPRS Baktimakmur API is Running", "version": "1.0.0", "env": config.get_settings().APP_ENV } @app.get("/health") def health_check(db: Session = Depends(get_db)): """ Health Check endpoint untuk memastikan API dan DB hidup. """ try: # Query sederhana untuk tes koneksi DB db.execute("SELECT 1") return {"status": "healthy", "database": "connected"} except Exception as e: raise HTTPException(status_code=503, detail=f"Database Unhealthy: {str(e)}") if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)

python main.py

INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit) INFO: Started reloader process [xxxxx] using StatReload INFO: Started server process [xxxxx] INFO: Waiting for application startup. ✅ Successfully connected to MySQL Database! INFO: Application startup complete.