Sesi 4: Business Logic

from sqlalchemy import Column, BigInteger, String, Date, Text, Numeric from app.models.base import TimestampMixin, Base class Customer(Base, TimestampMixin): __tablename__ = "customers" id = Column(BigInteger, primary_key=True, index=True) national_id = Column(String(30), unique=True, nullable=True, index=True) name = Column(String(150), nullable=False) birth_date = Column(Date, nullable=True) phone = Column(String(30), nullable=True) email = Column(String(150), nullable=True) address = Column(Text, nullable=True) business_type = Column(String(100), nullable=True) monthly_income = Column(Numeric(18, 2), nullable=True) assets_value = Column(Numeric(18, 2), nullable=True)

from sqlalchemy import Column, BigInteger, Integer, String, Date, Numeric, Enum, ForeignKey from sqlalchemy.orm import relationship from app.models.base import TimestampMixin, Base import enum class LoanStatus(str, enum.Enum): active = "active" settled = "settled" defaulted = "defaulted" cancelled = "cancelled" class Loan(Base, TimestampMixin): __tablename__ = "loans" id = Column(BigInteger, primary_key=True, index=True) customer_id = Column(BigInteger, ForeignKey("customers.id"), nullable=False) branch_id = Column(Integer, ForeignKey("branches.id"), nullable=False) product_name = Column(String(100), nullable=True) principal_amount = Column(Numeric(18, 2), nullable=False) profit_rate = Column(Numeric(5, 2), nullable=False) # Contoh: 12.5 untuk 12.5% tenor_months = Column(Integer, nullable=False) start_date = Column(Date, nullable=False) end_date = Column(Date, nullable=True) collateral_desc = Column(Text, nullable=True) status = Column(Enum(LoanStatus), default=LoanStatus.active) # ML Fields (diisi nanti oleh model prediksi) credit_score = Column(Numeric(6, 2), nullable=True) default_probability = Column(Numeric(5, 4), nullable=True) segment = Column(String(10), nullable=True) # 'low', 'medium', 'high' # Relationships customer = relationship("Customer", backref="loans") payments = relationship("Payment", back_populates="loan") class Payment(Base, TimestampMixin): __tablename__ = "payments" id = Column(BigInteger, primary_key=True, index=True) loan_id = Column(BigInteger, ForeignKey("loans.id"), nullable=False) schedule_date = Column(Date, nullable=False) paid_date = Column(Date, nullable=True) amount_due = Column(Numeric(18, 2), nullable=False) amount_paid = Column(Numeric(18, 2), default=0) status = Column(String(20), default="on_time") # on_time, late, default loan = relationship("Loan", back_populates="payments")

from pydantic import BaseModel, Field from datetime import date, datetime from decimal import Decimal from typing import Optional class CustomerBase(BaseModel): name: str = Field(..., min_length=3) phone: Optional[str] = None email: Optional[str] = None business_type: Optional[str] = None monthly_income: Optional[Decimal] = None assets_value: Optional[Decimal] = None class CustomerCreate(CustomerBase): national_id: Optional[str] = None class CustomerResponse(CustomerBase): id: int created_at: datetime class Config: from_attributes = True class LoanCreate(BaseModel): customer_id: int branch_id: int product_name: Optional[str] = None principal_amount: Decimal = Field(..., gt=0) profit_rate: Decimal = Field(..., ge=0) # Rate dalam persen (misal 15.5) tenor_months: int = Field(..., gt=0) start_date: date collateral_desc: Optional[str] = None class LoanResponse(BaseModel): id: int customer_id: int principal_amount: Decimal profit_rate: Decimal tenor_months: int monthly_installment: Optional[Decimal] = None status: str class Config: from_attributes = True

from sqlalchemy.orm import Session from datetime import date, timedelta from decimal import Decimal from app.models.loan import Loan, LoanStatus, Payment from app.models.customer import Customer from app.schemas.loan import LoanCreate def calculate_flat_rate(principal: Decimal, rate_percent: Decimal, tenor: int) -> Decimal: """ Menghitung angsuran bulanan dengan metode Flat Rate. Rumus: (Principal * (1 + (Rate/100))) / Tenor """ # Hitung total margin/profit total_profit = principal * (rate_percent / Decimal(100)) # Total utang + margin total_debt = principal + total_profit # Angsuran per bulan monthly = total_debt / Decimal(tenor) return monthly def create_loan_service(db: Session, loan_data: LoanCreate, current_user_id: int): # 1. Cek nasabah ada customer = db.query(Customer).filter(Customer.id == loan_data.customer_id).first() if not customer: raise ValueError("Customer not found") # 2. Hitung angsuran monthly_installment = calculate_flat_rate( loan_data.principal_amount, loan_data.profit_rate, loan_data.tenor_months ) # 3. Tentukan End Date (Start Date + Tenor Months) # Simple logic: Tambahkan bulan end_date = loan_data.start_date # Logic sederhana menambah bulan (bisa gunakan library dateutil relatifelta untuk lebih akurat) # Di sini kita pakai loop sederhana untuk clarity from dateutil.relativedelta import relativedelta end_date = loan_data.start_date + relativedelta(months=+loan_data.tenor_months) # 4. Buat Object Loan new_loan = Loan( **loan_data.dict(), end_date=end_date, status=LoanStatus.active ) db.add(new_loan) db.flush() # Flush agar dapat ID loan untuk membuat payments # 5. Buat Jadwal Pembayaran (Payments) # Generate schedule untuk setiap bulan current_sched_date = loan_data.start_date + relativedelta(months=+1) # Angsuran pertama bulan depan for i in range(loan_data.tenor_months): payment = Payment( loan_id=new_loan.id, schedule_date=current_sched_date, amount_due=monthly_installment, amount_paid=0, status="on_time" ) db.add(payment) current_sched_date = current_sched_date + relativedelta(months=+1) db.commit() db.refresh(new_loan) return new_loan

import os import pandas as pd from datetime import datetime from fastapi import UploadFile, HTTPException # Konfigurasi direktori upload (Pastikan folder ini ada di server) UPLOAD_DIR = "/www/wwwroot/app-collection/ao/uploads" os.makedirs(UPLOAD_DIR, exist_ok=True) async def save_upload_file(file: UploadFile): # Validasi tipe file if not (file.filename.endswith('.csv') or file.filename.endswith('.xlsx') or file.filename.endswith('.xls')): raise HTTPException(status_code=400, detail="Only CSV or Excel files are allowed") # Buat nama file unik: timestamp_filename.ext file_location = f"{UPLOAD_DIR}/{int(datetime.now().timestamp())}_{file.filename}" # Simpan file ke disk try: with open(file_location, "wb+") as file_object: file_object.write(file.file.read()) except Exception as e: raise HTTPException(status_code=500, detail=f"Could not save file: {str(e)}") # Baca file untuk preview summary try: if file.filename.endswith('.csv'): df = pd.read_csv(file_location) else: df = pd.read_excel(file_location) return { "filename": file.filename, "stored_path": file_location, "rows_count": len(df), "columns": list(df.columns), "preview": df.head(5).to_dict(orient="records") } except Exception as e: # Jika file terbaca error tapi file tersimpan return { "filename": file.filename, "stored_path": file_location, "error": "File saved but could not parse content for preview", "details": str(e) }

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session from typing import List from db import get_db from app.models.customer import Customer from app.schemas.loan import CustomerCreate, CustomerResponse from app.api.deps import get_current_active_user router = APIRouter() @router.post("/", response_model=CustomerResponse) def create_customer( customer: CustomerCreate, db: Session = Depends(get_db), current_user = Depends(get_current_active_user) # Harus login ): # Cek NIK duplikat jika diisi if customer.national_id: existing = db.query(Customer).filter(Customer.national_id == customer.national_id).first() if existing: raise HTTPException(status_code=400, detail="National ID already registered") db_customer = Customer(**customer.dict()) db.add(db_customer) db.commit() db.refresh(db_customer) return db_customer @router.get("/", response_model=List[CustomerResponse]) def list_customers(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)): customers = db.query(Customer).offset(skip).limit(limit).all() return customers

from fastapi import APIRouter, Depends from sqlalchemy.orm import Session from db import get_db from app.schemas.loan import LoanCreate, LoanResponse from app.services.loan_service import create_loan_service from app.api.deps import get_current_active_user router = APIRouter() @router.post("/submit", response_model=LoanResponse) def submit_loan( loan_data: LoanCreate, db: Session = Depends(get_db), current_user = Depends(get_current_active_user) ): try: new_loan = create_loan_service(db, loan_data, current_user.id) return new_loan except ValueError as e: from fastapi import HTTPException raise HTTPException(status_code=404, detail=str(e)) except Exception as e: from fastapi import HTTPException raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")

from fastapi import APIRouter, Depends, UploadFile, File from app.services.file_service import save_upload_file from app.api.deps import get_current_active_user router = APIRouter() @router.post("/upload") async def upload_file( file: UploadFile = File(...), current_user = Depends(get_current_active_user) ): """ Upload CSV atau XLSX untuk prediksi massal. File akan disimpan di server dan dikembalikan ringkasan isinya. """ result = await save_upload_file(file) return result

from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware from db import engine import config # Import Routers from app.api.auth import router as auth_router from app.api.customers import router as cust_router from app.api.loans import router as loan_router from app.api.files import router as file_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.include_router(cust_router, prefix="/api/v1/customers", tags=["Customers"]) app.include_router(loan_router, prefix="/api/v1/loans", tags=["Loans"]) app.include_router(file_router, prefix="/api/v1/files", tags=["Files & Uploads"]) @app.on_event("startup") def on_startup(): try: with engine.connect() as conn: print("✅ DB Connected!") # Buat folder uploads jika belum ada import os os.makedirs("/www/wwwroot/app-collection/ao/uploads", exist_ok=True) print("✅ Uploads directory checked/created.") except Exception as e: print(f"❌ Error: {e}") @app.get("/") def root(): return {"status": "BPRS API Running", "session": "4 - Business Logic Active"}