coding tutorial

FAST API Simple CRUD

FastAPI adalah framework Python modern yang dirancang untuk membangun API dengan cepat, efisien, dan mudah dikelola. Artikel ini akan membahas langkah-langkah sederhana untuk membuat aplikasi CRUD (Create, Read, Update, Delete) menggunakan FastAPI, yang cocok untuk pemula maupun pengembang berpengalaman.

app requirement

Database & Languange Programming

postgresql min v14
python min v3.10

install postgresql dengan docker image

docker run -d \
  --name postgres \
  -e POSTGRES_USER=root \
  -e POSTGRES_PASSWORD=admin1 \
  -e POSTGRES_DB=my_database \
  -v /home/learn_docker/postgresql/data:/var/lib/postgresql/data \
  -p 5432:5432 \
  postgres:14

project structure

/project_root
│
├── /simple_api
│   └── /app
│       └── main.py
├── /.venv
└── requirements.txt

pada directory project_root install virtual environment dengan nama .venv
kemudian aktifkan virtual environemtnya

# change directoty
cd path/to/project_root

# install environment
python -m venv .venv

# activate environment
source .venv/bin/activate

requirement.txt

annotated-types==0.7.0
anyio==4.8.0
click==8.1.8
exceptiongroup==1.2.2
fastapi==0.115.6
greenlet==3.1.1
h11==0.14.0
idna==3.10
psycopg2-binary==2.9.10
pydantic==2.10.4
pydantic_core==2.27.2
sniffio==1.3.1
starlette==0.41.3
typing_extensions==4.12.2
uvicorn==0.34.0

install library berdasarkan file requirements.txt

cd path/to/project_root

# activate environment
source .venv/bin/activate

pip install -r requirements.txt

atau install library langsung kemudian perbarui file requirements.txt

cd path/to/project_root

# activate environment
source .venv/bin/activate

pip install fastapi uvicorn psycopg2-binary pydantic

pip freeze > requirements.txt

main.py

''' import statement '''
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, condecimal
import psycopg2
from psycopg2.extras import RealDictCursor
from typing import List, Optional

''' import statement posgresql'''
SQLALCHEMY_DATABASE_URL = "postgresql://root:admin1@localhost/ecommerce"  # Ganti dengan kredensial PostgreSQL Anda

''' FastAPI setup '''
app = FastAPI()

''' Pydantic model untuk input dan output '''
class ProductCreate(BaseModel):
    name: str
    description: str
    price: condecimal(gt=0)  # harga harus lebih besar dari 0

class Product(ProductCreate):
    id: int

''' 
Database connection
Fungsi untuk mendapatkan koneksi ke database
'''
def get_db():
    conn = psycopg2.connect(SQLALCHEMY_DATABASE_URL)
    return conn

''' CRUD operation '''
@app.post("/products/", response_model=Product)
def create_product(product: ProductCreate):
    conn = get_db()
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cursor.execute(
            "INSERT INTO products (name, description, price) VALUES (%s, %s, %s) RETURNING id, name, description, price",
            (product.name, product.description, product.price)
        )
        new_product = cursor.fetchone()
        conn.commit()
        return new_product
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

@app.get("/products/", response_model=List[Product])
def get_products():
    conn = get_db()
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cursor.execute("SELECT id, name, description, price FROM products")
        products = cursor.fetchall()
        return products
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int):
    conn = get_db()
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cursor.execute("SELECT id, name, description, price FROM products WHERE id = %s", (product_id,))
        product = cursor.fetchone()
        if product is None:
            raise HTTPException(status_code=404, detail="Product not found")
        return product
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

@app.put("/products/{product_id}", response_model=Product)
def update_product(product_id: int, product: ProductCreate):
    conn = get_db()
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cursor.execute(
            "UPDATE products SET name = %s, description = %s, price = %s WHERE id = %s RETURNING id, name, description, price",
            (product.name, product.description, product.price, product_id)
        )
        updated_product = cursor.fetchone()
        if updated_product is None:
            raise HTTPException(status_code=404, detail="Product not found")
        conn.commit()
        return updated_product
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

@app.delete("/products/{product_id}", response_model=Product)
def delete_product(product_id: int):
    conn = get_db()
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    try:
        cursor.execute("SELECT id, name, description, price FROM products WHERE id = %s", (product_id,))
        product = cursor.fetchone()
        if product is None:
            raise HTTPException(status_code=404, detail="Product not found")
        
        cursor.execute("DELETE FROM products WHERE id = %s RETURNING id, name, description, price", (product_id,))
        deleted_product = cursor.fetchone()
        conn.commit()
        return deleted_product
    except Exception as e:
        conn.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        cursor.close()
        conn.close()

app run

cd path/to/project_root/simple_api

uvicorn main:app --reload

code explanation

Import statement

from fastapi import FastAPI, HTTPException

FastAPI: Framework utama yang digunakan untuk membuat aplikasi web dan API.
HTTPException: Digunakan untuk menangani dan mengembalikan error dengan status HTTP tertentu.

from pydantic import BaseModel, condecimal

BaseModel: Kelas dasar dari Pydantic yang digunakan untuk memvalidasi data input/output.
condecimal: Validator yang memastikan nilai numerik memiliki batasan tertentu, seperti harus lebih besar dari nol (gt=0).

import psycopg2
from psycopg2.extras import RealDictCursor

psycopg2: Library Python untuk berkomunikasi dengan PostgreSQL.
RealDictCursor: Cursor yang mengembalikan hasil query sebagai dictionary, sehingga lebih mudah diakses dengan nama kolom.

from typing import List, Optional

List: Digunakan untuk mendefinisikan tipe data list pada response.
Optional: Menandakan bahwa sebuah field bersifat opsional.

Database Setup

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

#URL koneksi untuk database PostgreSQL. Formatnya adalah
#postgresql://<username>:<password>@<host>/<database_name>

FastAPI App

# Membuat instance dari aplikasi FastAPI.
app = FastAPI()

Pydantic Models

class ProductCreate(BaseModel):
    name: str
    description: str
    price: condecimal(gt=0)
    
class Product(ProductCreate):
    id: int

ProductCreate: Model untuk memvalidasi data input (saat membuat atau mengupdate produk).
price: condecimal(gt=0): Memastikan bahwa harga harus lebih besar dari 0.
Product: Model output untuk produk, yang mewarisi ProductCreate dan menambahkan field id.

Database Connection

# Fungsi untuk mendapatkan koneksi ke database PostgreSQL.
def get_db():
    conn = psycopg2.connect(SQLALCHEMY_DATABASE_URL)
    return conn

CRUD Operations

create product

@app.post("/products/", response_model=Product)
def create_product(product: ProductCreate):

Endpoint: /products/ (HTTP POST) untuk membuat produk baru.
response_model=Product: Mengembalikan data produk yang baru dibuat dalam format yang sesuai dengan model Product.

# Query SQL untuk memasukkan data baru ke tabel products.
cursor.execute(
    "INSERT INTO products (name, description, price) VALUES (%s, %s, %s) RETURNING id, name, description, price",
    (product.name, product.description, product.price)
)

get all products

@app.get("/products/", response_model=List[Product])
def get_products():

Endpoint: /products/ (HTTP GET) untuk mendapatkan semua produk.
response_model=List[Product]: Mengembalikan daftar produk dalam format list dari model Product.

# Query SQL untuk mengambil semua data produk dari tabel.
cursor.execute("SELECT id, name, description, price FROM products")
products = cursor.fetchall()

get single product

@app.get("/products/{product_id}", response_model=Product)
def get_product(product_id: int):

Endpoint: /products/{product_id} (HTTP GET) untuk mendapatkan satu produk berdasarkan ID.
response_model=Product: Mengembalikan detail produk dalam format model Product.

# Query SQL untuk mengambil satu produk berdasarkan ID.
cursor.execute("SELECT id, name, description, price FROM products WHERE id = %s", (product_id,))

Update product

@app.put("/products/{product_id}", response_model=Product)
def update_product(product_id: int, product: ProductCreate):

Endpoint: /products/{product_id} (HTTP PUT) untuk mengupdate produk berdasarkan ID.
response_model=Product: Mengembalikan data produk yang baru saja diperbarui.

# Query SQL untuk memperbarui produk berdasarkan ID.
cursor.execute(
    "UPDATE products SET name = %s, description = %s, price = %s WHERE id = %s RETURNING id, name, description, price",
    (product.name, product.description, product.price, product_id)
)

Delete product

@app.delete("/products/{product_id}", response_model=Product)
def delete_product(product_id: int):

Endpoint: /products/{product_id} (HTTP DELETE) untuk menghapus produk berdasarkan ID.
response_model=Product: Mengembalikan data produk yang dihapus.

# Query SQL untuk menghapus produk dari tabel berdasarkan ID.
cursor.execute("DELETE FROM products WHERE id = %s RETURNING id, name, description, price", (product_id,))

Error handling

'''
Digunakan untuk mengembalikan error 404 jika produk tidak ditemukan.
'''
raise HTTPException(status_code=404, detail="Product not found")

'''
Jika ada error selama operasi database, rollback dilakukan, dan error dikembalikan ke klien dengan status HTTP 400.
'''
except Exception as e:
    conn.rollback()
    raise HTTPException(status_code=400, detail=str(e))

Connection cleanup

finally:
    cursor.close()
    conn.close()

finally: Menutup cursor dan koneksi database untuk memastikan tidak ada kebocoran sumber daya.

pydantic library

Pydantic adalah sebuah library Python yang digunakan untuk memvalidasi data dan mendefinisikan skema data menggunakan tipe data Python. Library ini sangat populer karena dapat memastikan bahwa data yang diterima atau digunakan dalam aplikasi sesuai dengan spesifikasi yang diinginkan, serta secara otomatis mengonversi tipe data jika diperlukan.

Keuntungan Menggunakan Pydantic:

Produktivitas Tinggi: Mempermudah validasi data yang kompleks dengan sedikit kode.
Integrasi API: Sangat cocok untuk digunakan dengan framework seperti FastAPI.
Efisien: Validasi data yang cepat tanpa kehilangan fleksibilitas.
Kompatibilitas JSON: Mendukung serialisasi/deserialisasi dengan mudah.

Fitur Utama Pydantic:

Validasi Data Otomatis:
- Pydantic memvalidasi data berdasarkan tipe anotasi Python (int, str, float, dll.).
- Jika data tidak valid, akan menghasilkan error dengan detail yang jelas.
Konversi Data Otomatis:
- Pydantic mencoba mengonversi data secara otomatis jika memungkinkan.
- Contohnya, string "42" bisa dikonversi menjadi integer 42.
Definisi Model yang Jelas:
- Menggunakan class berbasis Python untuk mendefinisikan skema data.
Error Handling yang Mudah:
- Pydantic memberikan pesan error yang deskriptif saat data tidak sesuai dengan skema.
Kompatibilitas dengan JSON dan API:
- Pydantic sering digunakan bersama framework seperti FastAPI untuk memvalidasi dan memproses input API berbasis JSON.

contoh penggunaan

definisi model dasar

from pydantic import BaseModel

class User(BaseModel):
    id: int
    name: str
    is_active: bool = True  # Default value

# Membuat instance model
user = User(id=1, name="John Doe")
print(user)
# Output: id=1 name='John Doe' is_active=True

validasi data

from pydantic import BaseModel, ValidationError

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

try:
    user = User(id="not-an-integer", name="John Doe")
except ValidationError as e:
    print(e)
# Output: 1 validation error for User
# id
#   value is not a valid integer (type=type_error.integer)

konversi tipe data

user = User(id="42", name="Alice")
print(user)
# Output: id=42 name='Alice'

Model Bersarang

class Address(BaseModel):
    city: str
    postal_code: str

class User(BaseModel):
    id: int
    name: str
    address: Address

user = User(id=1, name="John", address={"city": "New York", "postal_code": "10001"})
print(user)
# Output: id=1 name='John' address=Address(city='New York', postal_code='10001')

Validasi Costum

from pydantic import BaseModel, validator

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

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

try:
    user = User(id=1, name="John", age=-5)
except ValidationError as e:
    print(e)
# Output: 1 validation error for User
# age
#   Age must be a positive number (type=value_error)

lainnya

cursor.fetchone()
- Mengambil satu baris hasil query dari database.
- Jika tidak ada lagi baris yang tersedia, akan mengembalikan None.
cursor.fetchall()
- Mengambil semua baris hasil query dari database.
- Biasanya digunakan jika Anda ingin mengolah semua data sekaligus.
conn.cursor(cursor_factory=RealDictCursor)
- Digunakan untuk membuat objek cursor dari koneksi database (conn), dengan opsi cursor_factory untuk menentukan bagaimana hasil query akan dikembalikan.
- RealDictCursor adalah tipe cursor yang mengembalikan hasil query sebagai dictionary (dictionary Python). Setiap baris hasil query akan menjadi dictionary dengan nama kolom sebagai kunci dan nilai kolom sebagai nilai.
conn.commit()
- Digunakan untuk menyimpan semua perubahan yang dibuat selama sesi database ke dalam database.
- Biasanya digunakan setelah operasi seperti INSERT, UPDATE, atau DELETE agar perubahan permanen.
conn.rollback()
- Digunakan untuk membatalkan semua perubahan yang dibuat selama sesi transaksi jika terjadi error.
- Berguna dalam situasi di mana Anda tidak ingin perubahan disimpan ke database.