FastApi

Nestero 07 January 2025 | 15 January 2025

FastApi adalah salah satu frameworks python yang diperuntukan untuk membangun sebuah restapi, pada catatan ini akan dibahas sedikit tentang membuat sebuah endpoint, httpexception, database menggunakan sqlite.

Sebelum lebih lanjut persiapkan dahulu module atau library yang di butuhkan :

fastapi
uvicorn
fastapi_utils
sqlalchemy

simpan kedalam file requirements.txt dan install menggunakan pip

pip install -r requirements.txt

Database

File pertama yang akan dibuat adalah database.py, file database.py berisi settingan letak dan nama database yang akan digunakan.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from os import path

base_dir = path.dirname(path.abspath(__file__))
db_path = path.abspath(path.join(base_dir, "database.db"))
db_url = f"sqlite:///{db_path}"

engine = create_engine(db_url)

session_local = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_db():
    db = session_local()
    try:
        yield db
    finally:
        db.close()

setelah berhasil membuat file database selanjutnya adalah membuat model dan schema.

Model dan Schema

Model

Buat file models.py, dalam file models.py berisi settingan nama tabel dan isi kolom pada tabel beserta nilai kolom tabelnya.

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()


class Kontak(Base):
    __tablename__ = "kontak"
    id = Column(Integer, primary_key=True, index=True)
    nama = Column(String)
    no_tlp = Column(String)
    alamat = Column(String)
    tanggal_input = Column(DateTime)

Schema

Buat file schemas.py, dalam file schemas.py berisi skema yang akan digunakan untuk menginput atau mereturn data dalam database melalui endpoint yang akan dibuat nantinya.

from pydantic import BaseModel
from datetime import datetime


class KontakBase(BaseModel):
    nama: str
    alamat: str
    no_tlp: str
    tanggal_input: datetime


class KontakCreate(KontakBase):
    pass


class Kontak(KontakBase):
    id: int

    class Config:
        from_attributes = True

Endpoint

Setelah semua komponen file sudah siap dari database, model, dan schema, saatnya membuat endpoint, endpoint yang akan dibuat disini ada post, get, put, dan delete. Buat file dengan nama kontak.py, nama lain juga boleh bebas ^^.

from fastapi import Depends, HTTPException, APIRouter, Path
from sqlalchemy.orm import Session
from .database import get_db
from .schemas import Kontak, KontakCreate
from .models import Kontak as KontakModel
from sqlalchemy.sql import func

router = APIRouter()


# menambahkan kontak
@router.post("/", response_model=Kontak)
def add_kontak(data_kontak: KontakCreate, db: Session = Depends(get_db)):
    try:
        db_kontak = KontakModel(**data_kontak.model_dump())
        db.add(db_kontak)
        db.commit()
        db.refresh(db_kontak)
        return db_kontak
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


# list semua kontak
@router.get("/", response_model=dict)
def list_kontak(db: Session = Depends(get_db)):
    all_kontak = db.query(KontakModel).order_by(KontakModel.nama.asc()).all()
    total_kontak = db.query(func.count(KontakModel.id)).scalar()
    kontak_list = [Kontak.from_orm(kontak) for kontak in all_kontak]
    return {"total_kontak": total_kontak, "list_kontak": kontak_list}


# update kontak
@router.put("/{id_kontak}", response_model=dict)
def update_kontak(
    id_kontak: int = Path(..., description="id kontak yang akan diupdate"),
    data_kontak: KontakCreate | None = None,
    db: Session = Depends(get_db),
):
    kontak = db.query(KontakModel).filter(KontakModel.id == id_kontak).first()
    if kontak is None:
        raise HTTPException(status_code=404, detail="Kontak tidak ada di database")
    update_kontak = data_kontak.dict(exclude_unset=True)
    for key, value in update_kontak.items():
        setattr(kontak, key, value)
    db.commit()
    db.refresh(kontak)
    return {"msg": "kontak berhasil diupdate"}


# delete kontak
@router.delete("/{id_kontak}", response_model=dict)
def del_kontak(
    id_kontak: int = Path(..., description="id kontak yang akan dihapus"),
    db: Session = Depends(get_db),
):
    kontak = db.query(KontakModel).filter(KontakModel.id == id_kontak).first()
    if kontak is None:
        raise HTTPException(status_code=404, detail="Kontak tidak ada di database")
    db.delete(kontak)
    db.commit()
    return {"msg": "kontak berhasil dihapus"}

Main

Langkah terakhir membuat file main.py yang berisikan konfigurasi fastapinya.

from . import kontak, models
from .database import engine
from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

models.Base.metadata.create_all(bind=engine)
app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/")
def root():
    return {"msg": "hello ^^"}


app.include_router(kontak.router, tags=["Kontak"], prefix="/api/kontak")

Buat file __init__.py kosongkan saja, langkah terakhir ketikan perintah fastapi dev main.py diterminal dan akses localhost:8000/docs untuk melihat dokumentasi swagger yang otomatis dibuatkan oleh fastapi.