codingtutorial

FAST API CRUD (best practice)

Membangun operasi CRUD (Create, Read, Update, Delete) dengan FastAPI memerlukan pendekatan yang efisien dan terstruktur untuk memastikan performa dan kemudahan pengelolaan. Dalam artikel ini, kita akan membahas best practice seperti penggunaan Dependency Injection, validasi data dengan Pydantic, dan pengorganisasian kode untuk menciptakan aplikasi yang scalable dan maintainable.

Best Practice

A. Gunakan Dependency Injection

Dependency Injection (DI) di FastAPI adalah mekanisme untuk menyediakan dan mengelola dependensi (komponen eksternal atau fungsi) yang dibutuhkan oleh suatu fungsi atau endpoint secara otomatis. Dengan DI, Anda dapat memisahkan logika aplikasi dari pengelolaan sumber daya atau komponen tertentu, seperti database, layanan eksternal, atau konfigurasi aplikasi, sehingga kode menjadi lebih modular, bersih, dan mudah diuji.

Bagaimana Dependency Injection Bekerja di FastAPI

FastAPI memungkinkan Anda mendefinisikan dependensi menggunakan fungsi Python biasa atau class, kemudian mendaftarkannya sebagai argumen di endpoint. FastAPI secara otomatis akan memanggil fungsi tersebut, mengelola siklus hidupnya, dan menginjeksi hasilnya ke endpoint.

Berikut adalah beberapa contoh penggunaan Dependency Injection (DI) di FastAPI:

1. Mengelola Konfigurasi Aplikasi

Gunakan DI untuk menyediakan konfigurasi global ke endpoint.

from fastapi import Depends, FastAPI

app = FastAPI()

# Fungsi dependensi
def get_config():
    return {"app_name": "My FastAPI App", "version": "1.0.0"}

# Endpoint menggunakan dependensi
@app.get("/")
def read_root(config: dict = Depends(get_config)):
    return {"message": f"Welcome to {config['app_name']}!"}

2. Mengelola Koneksi Database

Gunakan DI untuk menyambungkan ke database, mengelola koneksi, dan memastikan koneksi ditutup setelah selesai.

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from database import SessionLocal

app = FastAPI()

# Fungsi dependensi
def get_db():
    db = SessionLocal()  # Membuka koneksi database
    try:
        yield db  # Mengembalikan instance database
    finally:
        db.close()  # Menutup koneksi database

# Endpoint menggunakan koneksi database
@app.get("/users/")
def read_users(db: Session = Depends(get_db)):
    users = db.query(User).all()  # Query data pengguna
    return users

3. Autentikasi API Key

Gunakan DI untuk memvalidasi API key yang dikirimkan oleh klien.

def get_api_key():
    return "my-secret-key"

def get_service(api_key: str = Depends(get_api_key)):
    return f"Service initialized with API key: {api_key}"

@app.get("/service/")
def read_service(service: str = Depends(get_service)):
    return {"service": service}

4. Dependensi Bersarang

Dependensi dapat saling memanggil untuk menyusun logika yang kompleks.

def get_api_key():
    return "my-secret-key"

def get_service(api_key: str = Depends(get_api_key)):
    return f"Service initialized with API key: {api_key}"

@app.get("/service/")
def read_service(service: str = Depends(get_service)):
    return {"service": service}

5. Logging atau Middleware Khusus

Gunakan DI untuk menyediakan logger atau middleware lainnya ke endpoint.

from fastapi import Depends, FastAPI

app = FastAPI()

# Fungsi dependensi untuk logger
def get_logger():
    return {"log_level": "info", "service": "FastAPI App"}

@app.get("/log/")
def read_log(logger: dict = Depends(get_logger)):
    return {"message": f"Logger is set to {logger['log_level']}"}

6. Pengelolaan Token JWT

Gunakan DI untuk memvalidasi dan memparsing token JWT.

from fastapi import Depends, FastAPI, HTTPException
import jwt

app = FastAPI()

# Fungsi untuk memvalidasi token
def get_current_user(token: str):
    try:
        payload = jwt.decode(token, "my-secret-key", algorithms=["HS256"])
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

@app.get("/profile/")
def read_profile(user: dict = Depends(get_current_user)):
    return {"user": user}

B. Definisikan Pydantic Models dengan Validasi

Pydantic adalah sebuah library Python yang digunakan untuk memvalidasi data dan mendefinisikan skema data menggunakan tipe data Python. Library ini mempermudah validasi dan konversi data secara otomatis berdasarkan anotasi tipe, sehingga cocok untuk digunakan dalam pengembangan aplikasi berbasis data atau API.

Mengapa Pydantic ?

  1. Validasi Data Otomatis:
    • Input dari klien (seperti JSON) divalidasi secara otomatis berdasarkan skema yang didefinisikan dengan Pydantic.
    • FastAPI memanfaatkan kemampuan Pydantic untuk memastikan data yang diterima valid sebelum diproses.
  2. Definisi Skema yang Jelas:
    • Skema data didefinisikan menggunakan class Python berbasis BaseModel dari Pydantic.
    • Skema ini juga digunakan untuk mendokumentasikan API secara otomatis menggunakan OpenAPI (Swagger).
  3. Konversi Data Otomatis:
    • Pydantic secara otomatis mengonversi tipe data saat diperlukan (misalnya, string menjadi integer).
    • Hal ini mengurangi kode boilerplate untuk parsing dan validasi data.
  4. Reusabilitas dan Konsistensi:
    • Model Pydantic dapat digunakan kembali di berbagai endpoint atau komponen aplikasi.
    • Membantu menjaga konsistensi data di seluruh aplikasi.
  5. Error Handling yang Terintegrasi:
    • Pydantic menghasilkan error yang terstruktur saat data tidak valid, dan FastAPI menggunakan error ini untuk memberikan respons HTTP 422 dengan detail kesalahan yang lengkap.
  6. Kompatibilitas dengan OpenAPI dan Swagger:
    • Skema Pydantic secara otomatis diintegrasikan dengan dokumentasi API yang dihasilkan oleh FastAPI.
    • Hal ini membantu pengembang dan pengguna API memahami input/output yang diperlukan.

1. Validasi Input

Gunakan model Pydantic untuk mendefinisikan struktur data yang harus diterima oleh endpoint.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    id: int
    name: str
    email: str

@app.post("/users/")
def create_user(user: User):
    return {"message": f"User {user.name} has been created!"}
  • Input JSON akan divalidasi berdasarkan model User.
  • Jika data tidak sesuai (contoh: id adalah string), FastAPI akan otomatis memberikan error respons.

2. Output Model

Gunakan model Pydantic untuk menentukan struktur data yang dikembalikan oleh endpoint.

class UserResponse(BaseModel):
    id: int
    name: str

@app.get("/users/{user_id}", response_model=UserResponse)
def get_user(user_id: int):
    return {"id": user_id, "name": "John Doe", "extra_field": "ignored"}
  • Field tambahan seperti extra_field akan diabaikan karena tidak sesuai dengan model UserResponse.

3. Validasi Kustom

Tambahkan validasi khusus menggunakan decorator @validator di Pydantic.

from pydantic import BaseModel, validator

class User(BaseModel):
    name: str
    age: int

    @validator("age")
    def validate_age(cls, value):
        if value < 0:
            raise ValueError("Age must be a positive number")
        return value

@app.post("/validate-user/")
def validate_user(user: User):
    return {"message": f"User {user.name} is valid."}

B. Menangani Error dan Exception

FastAPI memungkinkan Anda untuk menangani error dan exception secara eksplisit. Ini membantu untuk memberikan umpan balik yang jelas kepada pengguna ketika terjadi kesalahan.

from fastapi import HTTPException

@app.get("/items/{item_id}")
def get_item(item_id: int):
    item = get_item_from_db(item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

C. Gunakan BackgroundTasks untuk Proses Asinkron

Untuk pekerjaan yang memerlukan waktu lama (misalnya pengiriman email atau pemrosesan data yang berat), gunakan BackgroundTasks. Ini memungkinkan Anda untuk menangani tugas-tugas tersebut di background tanpa membuat pengguna menunggu.

from fastapi import BackgroundTasks

def send_email(email: str):
    # Kode untuk mengirim email
    pass

@app.post("/send-notification/")
def send_notification(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email)
    return {"message": "Notification sent in the background"}

D. Gunakan Middlewares untuk Pengelolaan Global

Middlewares dapat digunakan untuk menangani proses yang berlaku untuk setiap request, seperti logging, CORS, autentikasi, atau pengolahan header.

from fastapi import FastAPI

app = FastAPI()

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

E. Membuat Dokumentasi API

FastAPI secara otomatis menghasilkan dokumentasi API yang sangat berguna menggunakan OpenAPI dan Swagger UI. Anda tidak perlu menulis dokumentasi manual, karena FastAPI melakukan ini untuk Anda. Gunakan endpoint /docs untuk melihat antarmuka Swagger yang interaktif.

@app.get("/items/")
def get_items():
    """
    Retrieve a list of items.
    Returns:
        List of items
    """
    return items

F. Penerapan CORS (Cross-Origin Resource Sharing)

Jika aplikasi Anda perlu diakses dari domain yang berbeda (misalnya front-end dan back-end terpisah), pastikan untuk mengonfigurasi CORS dengan benar.

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Atau domain yang lebih spesifik
    allow_credentials=True,
    allow_methods=["*"],  # Atau pilih metode HTTP yang dibolehkan
    allow_headers=["*"],
)

G. Gunakan Asynchronous Programming untuk Kecepatan

FastAPI mendukung pemrograman asinkron (asyncio). Gunakan async dan await untuk operasi I/O yang lama (seperti pemanggilan API eksternal atau akses database) agar aplikasi Anda lebih responsif dan efisien.

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = await get_item_from_db(item_id)  # Menggunakan operasi async
    return item

H. Pengujian dan Unit Testing

Pengujian adalah bagian penting dari setiap aplikasi. FastAPI terintegrasi dengan baik dengan framework pengujian seperti pytest. Anda bisa menulis pengujian untuk endpoint API dan logika bisnis aplikasi dengan mudah.

from fastapi.testclient import TestClient

client = TestClient(app)

def test_get_item():
    response = client.get("/items/1")
    assert response.status_code == 200
    assert response.json() == {"name": "Item 1", "price": 10.0}

I. Optimalisasi dan Keamanan

Beberapa aspek yang perlu diperhatikan untuk meningkatkan kinerja dan keamanan aplikasi FastAPI:

  • Gunakan HTTPS untuk enkripsi komunikasi.
  • Rate Limiting untuk membatasi jumlah request dari pengguna.
  • Autentikasi dan Otorisasi dengan OAuth2, JWT (JSON Web Tokens), atau Basic Auth.
  • Logging untuk memantau kinerja dan aktivitas API.
  • Validation untuk memastikan bahwa input yang diterima API valid dan aman.

J. Menyusun Kode dengan Struktur yang Bersih

Meskipun FastAPI memungkinkan Anda untuk menulis kode yang sangat sederhana, untuk aplikasi yang lebih besar, pastikan Anda mengikuti prinsip-prinsip desain perangkat lunak yang baik. Pisahkan aplikasi Anda ke dalam folder yang terstruktur dengan baik, misalnya:

  • /app/models/ untuk model Pydantic atau ORM.
  • /app/routers/ untuk definisi route.
  • /app/services/ untuk logika bisnis.
  • /app/utils/ untuk fungsi-fungsi utilitas.

Struktur Direktori

Berikut adalah struktur proyek FastAPI yang disarankan, terutama untuk aplikasi yang lebih besar atau yang memerlukan pengelolaan kode yang lebih baik. Struktur ini memisahkan kode berdasarkan fungsionalitas, sehingga lebih mudah untuk memelihara, mengembangkan, dan menguji aplikasi.

Struktur Direktori yang Direkomendasikan

/app
    ├── /api                   # Folder untuk router/endpoints
    │   ├── /v1                # Versi API (misalnya v1, v2, dst.)
    │   │   ├── __init__.py
    │   │   ├── endpoints.py   # Tempat definisi route
    │   │   └── users.py       # Route terkait "users"
    │   └── __init__.py
    ├── /core                  # Kode utama dan konfigurasi aplikasi
    │   ├── __init__.py
    │   ├── config.py          # Konfigurasi aplikasi (misalnya, database, JWT)
    │   ├── security.py        # Fungsi terkait autentikasi dan keamanan
    │   ├── exceptions.py      # Kode penanganan error umum
    │   └── utils.py           # Fungsi utilitas umum
    ├── /db                    # Kode terkait database (ORM, koneksi, dsb.)
    │   ├── __init__.py
    │   ├── base.py            # Kelas dasar untuk model ORM
    │   ├── models.py          # Definisi model database
    │   ├── session.py         # Fungsi terkait sesi database
    │   └── crud.py            # Operasi CRUD (Create, Read, Update, Delete)
    ├── /models                # Pydantic models untuk validasi
    │   ├── __init__.py
    │   ├── user.py            # Model Pydantic untuk User
    │   └── item.py            # Model Pydantic untuk Item
    ├── /services              # Logika bisnis dan komunikasi antar komponen
    │   ├── __init__.py
    │   ├── user_service.py    # Layanan yang berhubungan dengan pengguna
    │   └── item_service.py    # Layanan yang berhubungan dengan item
    ├── /tests                 # Folder untuk pengujian aplikasi
    │   ├── __init__.py
    │   ├── test_users.py      # Pengujian untuk endpoint user
    │   └── test_items.py      # Pengujian untuk endpoint item
    ├── /scripts               # Skrip untuk setup dan maintenance (misalnya migration, seed data)
    └── main.py                # Titik masuk aplikasi FastAPI

Deskripsi Folder dan File

/api

Folder ini berisi kode untuk router dan endpoint API. Anda bisa memisahkan route berdasarkan versi API atau fitur (misalnya, user, item, order). Setiap modul atau file bisa berisi route yang terkait dengan entitas tertentu, dan diimpor ke dalam file utama aplikasi.

  • /v1/endpoints.py: Menyimpan definisi route seperti GET, POST, PUT, atau DELETE untuk operasi terkait entitas tertentu.

/core

Folder ini berisi konfigurasi dan logika umum yang digunakan oleh seluruh aplikasi.

  • config.py: Tempat menyimpan pengaturan konfigurasi aplikasi seperti koneksi database, pengaturan JWT, dsb.
  • security.py: Berisi kode terkait autentikasi, otorisasi, dan validasi token (misalnya menggunakan OAuth2 atau JWT).
  • exceptions.py: Berisi kode untuk menangani pengecualian atau kesalahan yang mungkin muncul di aplikasi.
  • utils.py: Menyimpan fungsi-fungsi utilitas yang berguna secara umum dalam aplikasi.

/db

Berisi kode untuk pengelolaan database dan ORM (Object-Relational Mapping) jika Anda menggunakan database. Biasanya di sini Anda akan mendefinisikan model database dan CRUD operations.

  • models.py: Menyimpan model ORM untuk entitas seperti User, Item, dsb.
  • session.py: Kode untuk menangani sesi koneksi ke database.
  • crud.py: Menyimpan logika CRUD, misalnya menambahkan data, mendapatkan data, memperbarui, dan menghapus data.

/models

Berisi Pydantic models yang digunakan untuk validasi input/output API. Ini adalah tempat untuk mendefinisikan struktur data yang diterima dan dikirim oleh API.

  • user.py: Model Pydantic untuk validasi data pengguna.
  • item.py: Model Pydantic untuk validasi data item.

/services

Menampung logika bisnis yang lebih kompleks dan komunikasi antara model, database, dan API. Di sini Anda bisa mendefinisikan fungsi atau layanan yang menangani operasi lebih mendalam atau yang menghubungkan beberapa entitas.

  • user_service.py: Menyediakan fungsi terkait pengelolaan pengguna, seperti registrasi, login, atau perubahan data.
  • item_service.py: Menyediakan logika terkait pengelolaan item (misalnya menambah, mengedit, atau menghapus item).

/tests

Folder ini menyimpan unit tests dan integration tests untuk aplikasi Anda. Anda bisa menggunakan framework seperti pytest untuk menulis pengujian yang memastikan API Anda berfungsi dengan benar.

  • test_users.py: Pengujian untuk memastikan endpoint dan fungsionalitas terkait pengguna berjalan sesuai yang diharapkan.
  • test_items.py: Pengujian untuk memastikan endpoint dan fungsionalitas terkait item berjalan sesuai yang diharapkan.

/scripts

Tempat untuk skrip tambahan seperti migration database, seed data, atau penanganan tugas administratif lainnya.

main.py

Ini adalah titik masuk utama untuk aplikasi FastAPI Anda. Di sini, Anda akan mengimpor dan menyatukan router dari /api, konfigurasi dari /core, dan middleware atau pengaturan tambahan lainnya.

Kelebihan Struktur Ini

  • Modular: Aplikasi dipisahkan berdasarkan fungsionalitasnya (misalnya, API, database, service), sehingga lebih mudah dipelihara.
  • Skalabilitas: Dengan pembagian seperti ini, Anda dapat dengan mudah menambah fitur atau versi API baru tanpa mengganggu kode yang sudah ada.
  • Pengujian: Folder tests membantu Anda untuk menguji aplikasi dan memastikan bahwa kode yang baru ditambahkan tidak merusak fungsionalitas yang ada.
  • Kebersihan Kode: Memisahkan logika bisnis dan pengelolaan database ke dalam folder terpisah menjaga kode tetap bersih dan terorganisir.

struktur direktori multiple apps

/my_project
    ├── /app
    │   ├── /app1              # Aplikasi pertama
    │   │   ├── /api           # API routers untuk app1
    │   │   │   ├── __init__.py
    │   │   │   ├── products.py
    │   │   │   └── orders.py
    │   │   ├── /models        # Models untuk app1
    │   │   │   ├── __init__.py
    │   │   │   ├── product.py
    │   │   │   └── order.py
    │   │   ├── /services      # Service layer untuk app1
    │   │   │   ├── __init__.py
    │   │   │   ├── product_service.py
    │   │   │   └── order_service.py
    │   │   ├── main.py        # Titik masuk untuk app1
    │   │   └── config.py      # Konfigurasi khusus app1
    │   ├── /app2              # Aplikasi kedua
    │   │   ├── /api           # API routers untuk app2
    │   │   │   ├── __init__.py
    │   │   │   ├── users.py
    │   │   │   └── payments.py
    │   │   ├── /models        # Models untuk app2
    │   │   │   ├── __init__.py
    │   │   │   ├── user.py
    │   │   │   └── payment.py
    │   │   ├── /services      # Service layer untuk app2
    │   │   │   ├── __init__.py
    │   │   │   ├── user_service.py
    │   │   │   └── payment_service.py
    │   │   ├── main.py        # Titik masuk untuk app2
    │   │   └── config.py      # Konfigurasi khusus app2
    ├── /core                  # Kode inti yang digunakan bersama
    │   ├── __init__.py
    │   ├── config.py          # Konfigurasi bersama (misalnya database, JWT)
    │   ├── exceptions.py      # Exception handler global
    │   ├── security.py        # Fungsi keamanan dan autentikasi umum
    │   └── utils.py           # Utilitas umum yang digunakan di seluruh aplikasi
    ├── /db                    # Koneksi database dan model umum
    │   ├── __init__.py
    │   ├── models.py          # Model database umum, jika ada (misalnya User)
    │   ├── session.py         # Fungsi untuk mengelola session database
    │   ├── crud.py            # Operasi CRUD umum
    │   └── base.py            # Kelas base untuk model ORM
    ├── /tests                 # Folder pengujian
    │   ├── __init__.py
    │   ├── test_app1.py       # Pengujian untuk app1
    │   ├── test_app2.py       # Pengujian untuk app2
    ├── requirements.txt       # Dependensi Python (untuk kedua aplikasi)
    ├── Dockerfile             # Docker configuration (jika digunakan)
    ├── README.md              # Deskripsi proyek
    └── main.py                # Titik masuk utama (untuk menjalankan aplikasi)

4. Contoh Struktur Proyek Lebih Kompleks

Untuk proyek lebih besar, Anda bisa menambahkan folder lain seperti /docs untuk dokumentasi, /notifications untuk layanan terkait email/notifications, atau /background_tasks untuk task yang berjalan di background.

Study Case

Database & Languange Programming

  • postgresql min v14
  • python min v3.10

app structure

/project_root
├── /ecommerce
│   └── /app
│        ├── /api
│        │   └── /v1
│        │       ├── products.py 
│        │       └── .........py 
│        ├── /core
│        │   └── config.py
│        ├── /db
│        │   ├── base.py
│        │   ├── models.py
│        │   ├── session.py
│        │   └── crud.py
│        ├── /models
│        │   ├── product.py
│        │   └── ........py
│        ├── /services
│        │   ├── product_service.py
│        │   └── ......._service.py
│        └── main.py
├── /.venv
└── requirements.txt
  • jika ada endpoint tambahan selain product tinggal buat file endpoint, models, dan servicenya saja

requirements.txt

fastapi==0.115.6
psycopg2-binary==2.9.10
pydantic==2.10.4
pydantic_core==2.27.2
SQLAlchemy==2.0.36
uvicorn==0.34.0

app.api.v1

  • .products.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.models.product import ProductCreate, Product
from app.services.product_services import (
    create_product_service, 
    get_all_products_service, 
    get_single_product_service, 
    update_product_service, 
    delete_product_service
)
from app.db.session import get_db

router = APIRouter()

@router.post("/", response_model=Product)
def add_product(product: ProductCreate, db: Session = Depends(get_db)):
    return create_product_service(db, product)

@router.get("/", response_model=list[Product])
def list_products(db: Session = Depends(get_db)):
    return get_all_products_service(db)

@router.get("/{product_id}", response_model=Product)
def get_product(product_id: int, db: Session = Depends(get_db)):
    product = get_single_product_service(db, product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

@router.put("/{product_id}", response_model=Product)
def update_product(product_id: int, product: ProductCreate, db: Session = Depends(get_db)):
    update_product = update_product_service(db, product_id, product)
    if not update_product:
        raise HTTPException(status_code=404, detail="Product not found")
    return update_product

@router.delete("/{product_id}", response_model=Product)
def delete_product(product_id: int, db: Session = Depends(get_db)):
    delete_product = delete_product_service(db, product_id)
    if not delete_product:
        raise HTTPException(status_code=404, detail="Product not found")
    return delete_product

Kode tersebut adalah implementasi CRUD (Create, Read, Update, Delete) untuk mengelola data produk menggunakan FastAPI, dengan praktik terbaik seperti pemisahan logika layanan (service) dan penggunaan Dependency Injection. Berikut penjelasan singkatnya:

  1. Import Library dan Modul:
    • Mengimpor modul seperti APIRouter untuk membuat grup endpoint, Depends untuk Dependency Injection, dan HTTPException untuk menangani error.
    • Mengimpor model data (ProductCreate, Product) dan layanan (product_services) yang berisi logika CRUD.
    • Mengimpor get_db untuk Dependency Injection koneksi database.
  2. Router Setup:
    • router = APIRouter(): Membuat router untuk grup endpoint yang dapat dipisahkan dari file utama.
  3. Endpoint CRUD:
    • @router.post("/"): Endpoint untuk menambahkan produk baru dengan validasi data dari model ProductCreate dan menyimpan ke database menggunakan layanan create_product_service.
    • @router.get("/"): Endpoint untuk mendapatkan daftar semua produk menggunakan layanan get_all_products_service.
    • @router.get("/{product_id}"): Endpoint untuk mengambil detail produk berdasarkan product_id, dengan validasi keberadaan produk. Jika tidak ditemukan, akan mengembalikan error 404.
    • @router.put("/{product_id}"): Endpoint untuk memperbarui data produk berdasarkan product_id dengan data baru yang divalidasi melalui ProductCreate.
    • @router.delete("/{product_id}"): Endpoint untuk menghapus produk berdasarkan product_id. Jika produk tidak ditemukan, mengembalikan error 404.
  4. Dependency Injection:
    • db: Session = Depends(get_db): Setiap endpoint menggunakan koneksi database yang dikelola secara otomatis dengan get_db.
  5. Response Model:
    • response_model: Mengontrol struktur data output menggunakan model Pydantic (Product atau list[Product]).

app.core.config.py

DATABASE_URL = "postgresql://root:admin1@localhost/ecommerce"

Kode tersebut adalah string koneksi database untuk PostgreSQL. Artinya, aplikasi akan terhubung ke database PostgreSQL dengan:

  • Username: root
  • Password: admin1
  • Host: localhost (komputer lokal)
  • Database: ecommerce.

app.core.db

  • .base.py
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

digunakan dalam SQLAlchemy untuk membuat kelas dasar yang menjadi dasar semua model tabel database Anda.

Penjelasan:

  1. declarative_base:
    • Fungsi yang menghasilkan kelas dasar yang disebut Base.
    • Kelas ini digunakan untuk mendefinisikan model tabel di database (ORM – Object Relational Mapping).
  2. Base:
    • Semua model tabel database akan mewarisi Base.
    • Ini memungkinkan SQLAlchemy mengenali model Anda sebagai tabel database.
  • .crud.py
from sqlalchemy.orm import Session
from app.db.models import Product
from app.models.product import ProductCreate

# Fungsi untuk menambahkan produk baru
def create_product(db: Session, product: ProductCreate):
    db_product = Product(
        name=product.name,
        description=product.description,
        price=product.price,
        stock=product.stock,
    )
    db.add(db_product)
    db.commit()
    db.refresh(db_product)
    return db_product

# Fungsi untuk mendapatkan daftar semua produk
def get_products(db: Session):
    return db.query(Product).all()

# Fungsi untuk mendapatkan produk berdasarkan ID
def get_product_by_id(db: Session, product_id: int):
    return db.query(Product).filter(Product.id == product_id).first()

# Fungsi untuk mengupdate produk berdasarkan ID
def update_product(db: Session, product_id: int, product_data: ProductCreate):
    db_product = db.query(Product).filter(Product.id == product_id).first()
    if db_product:
        db_product.name         = product_data.name
        db_product.description  = product_data.description
        db_product.price        = product_data.price
        db_product.stock        = product_data.stock
        db.commit()
        db.refresh(db_product)
        return db_product
    return None

# Fungsi untuk menghapus produk berdasarkan ID
def delete_product(db: Session, product_id: int):
    db_product =  db.query(Product).filter(Product.id == product_id).first()
    if db_product:
        db.delete(db_product)
        db.commit()
        return db_product
    return None

Kode tersebut adalah implementasi fungsi-fungsi CRUD (Create, Read, Update, Delete) untuk mengelola data produk menggunakan SQLAlchemy. Berikut penjelasannya:

  1. Import:
    • Session: Objek sesi database untuk menjalankan query.
    • Product: Model tabel database untuk produk.
    • ProductCreate: Skema Pydantic untuk validasi input data produk.
  2. Fungsi CRUD:
    • create_product: Menambahkan produk baru ke database dengan data dari ProductCreate.
    • get_products: Mengambil semua produk dari tabel Product.
    • get_product_by_id: Mengambil produk tertentu berdasarkan product_id.
    • update_product: Memperbarui data produk berdasarkan product_id dan data baru dari ProductCreate.
    • delete_product: Menghapus produk dari database berdasarkan product_id.
  3. Operasi SQLAlchemy:
    • db.add(): Menambahkan data baru ke sesi database.
    • db.commit(): Menyimpan perubahan ke database.
    • db.refresh(): Memperbarui objek setelah disimpan ke database.
    • db.query().filter().first(): Melakukan pencarian berdasarkan kondisi.
    • db.delete(): Menghapus data dari database.
  • .models.py
from sqlalchemy import Column, Integer, String, Float
from app.db.base import Base

class Product(Base):
    __tablename__ = "products"

    id          = Column(Integer, primary_key=True, index=True)
    name        = Column(String, index=True)
    description = Column(String)
    price       = Column(Float)
    stock       = Column(Integer)

Fungsi dari kode tersebut adalah untuk mengelola koneksi database dalam aplikasi. Secara spesifik:

  1. Membuat Engine:
    • create_engine digunakan untuk menghubungkan aplikasi ke database berdasarkan DATABASE_URL.
  2. Membuat Sesi Database:
    • SessionLocal adalah kelas sesi untuk menjalankan query database secara terisolasi.
  3. Dependency Injection (get_db):
    • Fungsi get_db menyediakan sesi database ke endpoint FastAPI dan memastikan sesi ditutup setelah selesai, sehingga menghindari kebocoran koneksi.
  • .session.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import DATABASE_URL

engine = create_engine(DATABASE_URL)

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

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

Kode tersebut mengatur koneksi database menggunakan SQLAlchemy untuk aplikasi. Berikut penjelasannya:

Penjelasan Komponen Kode:

  1. create_engine:
    • Membuat engine koneksi ke database berdasarkan URL koneksi yang didefinisikan di DATABASE_URL.
    • DATABASE_URL adalah string koneksi database (misalnya, PostgreSQL atau SQLite).
  2. sessionmaker:
    • Digunakan untuk membuat SessionLocal, yaitu kelas sesi database yang digunakan untuk menjalankan operasi query.
    • Parameter:
      • autocommit=False: Menonaktifkan commit otomatis; semua perubahan harus disimpan secara manual dengan db.commit().
      • autoflush=False: Menonaktifkan flush otomatis; SQLAlchemy tidak akan mengirimkan perubahan ke database hingga db.commit() dipanggil.
      • bind=engine: Menghubungkan sesi dengan engine yang telah dibuat.
  3. get_db:
    • Fungsi Dependency Injection untuk mengelola sesi database.
    • Membuka sesi database (db = SessionLocal()), menyediakan sesi ini ke fungsi yang memerlukan (yield db), lalu memastikan sesi ditutup setelah selesai (db.close()).

Fungsi Kode:

  • Membuat Engine: Menginisialisasi koneksi ke database.
  • Membuat Sesi: SessionLocal memungkinkan operasi database dilakukan dalam sesi terisolasi.
  • Dependency Injection: get_db menyediakan sesi database untuk digunakan di endpoint FastAPI tanpa harus mengelola koneksi secara manual.

app.models

  • product.py
from pydantic import BaseModel

class ProductBase(BaseModel):
    name : str 
    description : str 
    price : float
    stock : int

class ProductCreate(ProductBase):
    pass

class Product(ProductBase):
    id : int

    class Config:
        orm_mode = True

Kode tersebut mendefinisikan model Pydantic untuk validasi data produk:

  1. ProductBase: Model dasar dengan atribut name, description, price, dan stock.
  2. ProductCreate: Digunakan untuk validasi input saat membuat produk baru, mewarisi dari ProductBase.
  3. Product: Menambahkan atribut id untuk output data, dengan orm_mode = True agar kompatibel dengan model ORM seperti SQLAlchemy.

app.services

  • product_services.ps
from sqlalchemy.orm import Session
from app.db.crud import (
    create_product, get_products, get_product_by_id, update_product, delete_product
) 
from app.models.product import ProductCreate

def create_product_service(db: Session, product: ProductCreate):
    return create_product(db, product)

def get_all_products_service(db: Session):
    return get_products(db)

def get_single_product_service(db: Session, product_id: int):
    return get_product_by_id(db, product_id)

def update_product_service(db: Session, product_id: int, product: ProductCreate):
    return update_product(db, product_id, product)

def delete_product_service(db: Session, product_id: int):
    return delete_product(db, product_id)

Fungsi dari kode tersebut adalah untuk membuat layer layanan (service layer) yang bertugas sebagai perantara antara logika bisnis (CRUD) dan endpoint API.

Penjelasan Fungsi:

  1. create_product_service:
    • Memanggil fungsi create_product untuk menambahkan produk baru ke database.
  2. get_all_products_service:
    • Memanggil fungsi get_products untuk mengambil semua data produk dari database.
  3. get_single_product_service:
    • Memanggil fungsi get_product_by_id untuk mengambil data satu produk berdasarkan product_id.
  4. update_product_service:
    • Memanggil fungsi update_product untuk memperbarui data produk berdasarkan product_id.
  5. delete_product_service:
    • Memanggil fungsi delete_product untuk menghapus data produk berdasarkan product_id.

Tujuan:

  • Memisahkan logika bisnis (CRUD) dari endpoint API.
  • Mempermudah pengelolaan, pengujian, dan pemeliharaan kode dengan struktur yang modular.

main.py

from fastapi import FastAPI
from app.api.v1.products import router as products_router
from app.db.base import Base
from app.db.session import engine

app = FastAPI()

# include routers
app.include_router(products_router, prefix="/v1/products", tags=["products"])

# initialize database
Base.metadata.create_all(bind=engine)

@app.get("/")
def read_root():
    return {"message": "welcome to the product API"}

Kode tersebut digunakan untuk membuat dan menginisialisasi aplikasi API dengan FastAPI, termasuk konfigurasi rute, pengaturan database, dan endpoint dasar. Berikut fungsinya:

Penjelasan Fungsi:

  1. FastAPI():
    • Membuat instance aplikasi FastAPI.
  2. app.include_router:
    • Menyertakan router products_router untuk menangani semua rute terkait produk.
    • prefix="/v1/products": Menentukan prefiks URL untuk endpoint produk (misalnya, /v1/products).
    • tags=["products"]: Menambahkan kategori dalam dokumentasi API.
  3. Base.metadata.create_all(bind=engine):
    • Menginisialisasi tabel-tabel database sesuai model yang didefinisikan di ORM SQLAlchemy.
    • engine: Objek koneksi database.
  4. Endpoint /:
    • Endpoint dasar yang mengembalikan pesan selamat datang ({"message": "welcome to the product API"}).

app run

cd path/to/project_root/

# install environment
python -m venv .venv

# activate environemnt
source .venv/bin/activate

# install library
pip install -r requirements.txt

cd ecommerce

uvicorn app.main:app --reload
Tags

Related Articles

FAST API Directory Structure

January 9, 2025

FAST API Simple CRUD

January 9, 2025

Docker Compose

December 27, 2024

Docker Persistent Storage

December 27, 2024
Back to top button

Table of Contents

→ Index