Tutorial FastAPI: CRUD dengan MySQL

Oktober 3, 2021 • Category: Tutorial • 19 min read
Featured Image

Hallo guys, pada kesempatan ini saya akan menulis tutorial CRUD sederhana menggunakan Fastapi dengan database MySql. Selain itu juga kita akan mempraktekkan bagaimana caranya membuat migrations dan data seed menggunakan alembic.

Table of Contents

Instalasi

Pertama-tama kita akan membuat virtual environment terlebih dahulu. Agar nantinya kita bisa memilih versi pada project lainnya.

virtualenv env

Setelah itu masuk kedalam virtual environment

source env/bin/activate

Package yang dibutuhkan dalam tutorial ini yaitu:

  • fastapi
  • alembic
  • pydantic
  • uvicorn
  • SQLAlchemy
  • pymsql
  • Faker

Silahkan instal satu persatu. Atau dengan cara yang lain, yaitu buat file requirements.txt lalu pastekan dibawah ini

alembic==1.7.3
asgiref==3.4.1
bson==0.5.10
click==8.0.1
fastapi==0.68.1
Faker==8.14.0
greenlet==1.1.1
h11==0.12.0
httptools==0.2.0
jsonify==0.5
Mako==1.1.5
MarkupSafe==2.0.1
pydantic==1.8.2
PyMySQL==1.0.2
python-dateutil==2.8.2
python-dotenv==0.19.0
PyYAML==5.4.1
pytest==6.2.5
requests==2.26.0
six==1.16.0
SQLAlchemy==1.4.25
starlette==0.14.2
typing-extensions==3.10.0.2
uvicorn==0.15.0
uvloop==0.16.0
watchgod==0.7
websockets==10.0

Eksekusi file tersebut dengan pip

pip install -r requirements.txt

Migration dan Seeding

Setelah kalian mempersiapkan package yang dibutuhkan, selanjutnya kita akan membuat migration dan seeding menggunakan alembic.

Pertama-tama kita harus menginisialisasikan alembic melalui terminal/cmd didalam direktori project kita.

alembic init alembic

Pada proses diatas akan membuat folder dengan nama alembic. Lalu terdapat file alembic.ini itu adalah tempat konfigurasi pada alembic. Struktur folder saat ini:

├── alembic
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
├── alembic.ini
└── requirements.txt

Buka file alembic.ini dengan editor kalian. Ubah pada line 53 atau cari sqlalchemy.url. Ubah valuenya menjadi konfigurasi milikmu.

Sebelumnya

sqlalchemy.url = driver://user:pass@localhost/dbname

Setelahnya

sqlalchemy.url = mysql+pymysql://USERNAME_DB:PASSWORD_DB@localhost:3306/fastapi_crud

Penjelasan: pada mysql+pymysql ini merupakan driver yang kita gunakan untuk komunikasi antara project kita dengan database mysql. Ubah USERNAME_DB, PASSWORD_DB sesuai akses yang kamu miliki. pada _fastapicrud merupakan nama database dalam project ini.

Lalu buat database di mysql.

CREATE DATABASE fastapi_crud;

Setelah database terbuat, kita akan membuat migration. Ketikkan perintah dibawah

alembic revision -m "create table pegawai"

Perintah diatas akan membuat sebuah file baru didalam folder alembic/versions Didalam file tersebut ada func upgrade() dan downgrade(). Apa kegunaan dari 2 fungsi tersebut? silahkan baca dokumentasinya untuk lebih lengkapnya ya.

Setelah itu buka file baru tersebut. Disini pada contoh saya file yang baru bernama a94e7a90e9fa_create_table_pegawai.py Buka file tersebut, ubah kode dibawah ini setelah baris kode depends_on = None

Sebelum

def upgrade():
    pass


def downgrade():
    pass

Sesudah

from faker import Faker

faker = Faker('id_ID')

def upgrade():
    pgw = op.create_table(
        'dt_pegawai',
        sa.Column('id_pegawai', sa.Integer, primary_key=True),
        sa.Column('nama_pegawai', sa.String(255), nullable=False),
        sa.Column('alamat_pegawai', sa.String(255)),
        sa.Column('ttl_pegawai', sa.Date()),
        sa.Column('telp_pegawai', sa.String(255)),
        sa.Column('email_pegawai', sa.String(255), unique=True)
    )
    op.bulk_insert(
        pgw,
        [{'nama_pegawai':faker.name(), 
                 'alamat_pegawai':faker.address(),
                 'ttl_pegawai':faker.date_of_birth(tzinfo=None, minimum_age=21, maximum_age=28),
                 'telp_pegawai':faker.phone_number(),
                 'email_pegawai':faker.email()
                  } for x in range(800)]
    )


def downgrade():
    op.drop_table('dt_pegawai')

Simpan file tersebut. Lalu jalankan perintah dibawah ini untuk melakukan migrating dan seeding ke database.

alembic upgrade head

Silahkan cek database milikmu apakah sudah ada atau belum tabel dt_pegawai beserta seed datanya.

Konfigurasi Database

Setelah kamu berhasil membuat migrate dan data seed. selanjutnya kita akan membuat file konfigurasi database. Loh bedanya apa dengan yang tadi? Pada konfigurasi ini dikhususkan untuk aplikasi kita, sedangkan yang tadi ialah untuk migrations dan seed data.

Kita akan membuat folder baru bernama config lalu didalamnya ada file database.py. Masukkan kode dibawah ini pada file config/database.py

# config/database.py
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import configparser

config = configparser.ConfigParser()
config.read('alembic.ini')

SQLALCHEMY_DATABASE_URL = config.get('alembic', 'sqlalchemy.url')

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

Base = declarative_base()

conn = engine.connect()

Penjelasan: Disini kita menggunakan configparser sebagai pembaca file berekstensi .ini yang ada pada file alembic.ini. Didalam alembic.ini terdapat url database kita maka dari itu kita memanggil filenya lalu pilih section alembic lalu optionsnya yaitu sqlalchemy.url

config = configparser.ConfigParser()
config.read('alembic.ini')

SQLALCHEMY_DATABASE_URL = config.get('alembic', 'sqlalchemy.url')

Sehingga variable SQLALCHEMY_DATABASE_URL berisi url koneksi database kita. Selanjutnya kita membuat engine dengan function _createengine yang berisikan url database tadi. Setelah itu baru kita membuat session dari function sessionmaker lalu mengkoneksikan dengan database.

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

conn = engine.connect()

Membuat Schemas

Setelah kamu sudah membuat konfigurasi database untuk aplikasi, selanjutnya kita akan membuat schemas. Kenapa perlu membuat schemas? Jadi jika saya simpulkan menurut pribadi schemas itu seperti json validasi, selain itu juga schemas juga bisa untuk membuat struktur json. Dengan begitu kita harus membuat schemas sesuai dengan kebutuhan front-end untuk struktur json yang akan dibuat. Schemas juga bertugas sebagai validasi json, entah itu minimal/maksimal string, lalu tipedata, lalu default valuenya apa, dll. Untuk lebih lanjut tentang schemas silahkan baca dokumentasi ini ya.

Kita akan membuat folder baru bernama schemas lalu didalamnya ada file pegawai.py. Masukkan kode dibawah ini pada file schemas/pegawai.py

# schemas/pegawai.py
from datetime import date
from pydantic import BaseModel, Field
from typing import List


class PegawaiSchema(BaseModel):
    nama_pegawai: str = Field(..., min_length=3, max_length=255)
    alamat_pegawai: str = Field(..., min_length=3, max_length=255)
    ttl_pegawai: date = Field(...)
    telp_pegawai: str = Field(..., min_length=3, max_length=255)
    email_pegawai: str = Field(..., min_length=3, max_length=50)

class Pegawai(PegawaiSchema):
    id_pegawai: int

# list pegawai API
class Pegawais(BaseModel):
    limit: int = Field(default=5)
    offset: int = Field(default=0)
    data: List[Pegawai]

Penjelasan: Pada kode diatas terdapat 3 class.

Yang pertama PegawaiSchema dengan parameter BaseModel diimport dari pydantic. Pada class pertama ini 5 atribut/variable. Ditiap-tiapnya memiliki tipe data yang berbeda. Yang menjadi sorotan saya ialah function Field(). Parameter pertama ialah yaitu default, jika parameter pertama default berisikan ... atau Field(...) maka itu berarti atribut/variable itu tidak boleh kosong. Sedangkan pada parameter kedua dan seterusnya kita bisa menyesuaikan kebutuhan sesuai parameter yang tersedia. Disini saya menggunakan min_length dan max_length yang dimana untuk membatasi inputan dari client.

Pada class yang kedua memiliki 1 atribut/variable dengan parameter class PegawaiSchema. Kenapa demikian? karena nanti ketika kita ingin menampilkan datanya kita hanya memanggil class kedua ini. Pada class kedua ini berisikan id dari pegawai. Dengan kata lain sebagai primary-key. Jika kita memanggil class ini maka otomatis sistem akan juga memanggil pada class pertama. Pada atribut/variable id ini bertipe data integer yang berarti khusus untuk value angka.

Pada class yang ketiga memiliki parameter BaseModel diimport dari pydantic. Bedanya apa dengan class yang pertama? pada class ini dikhususkan untuk menampilkan data saja. Dengan kata lain class ini sebagai pondasi struktur json yang akan ditampilkan/diterima oleh client. Sedangkan pada class pertama itu dikhususkan untuk operasi Create,Update,dan delete data. Pada class ini terdapat 3 variable/atribut yang memiliki tipe data yang berbeda. Pada offset dan limit ini digunakan untuk pagination, sehingga client tau posisi datanya ada dimana dan limit berapa. Pada variable/atribut data ini ialah isi dari data tunggal tersebut sesuai dengan idnya.

Untuk lebih tau parameter apa saja yang ada didalam function Field() silahkan baca dokumentasi ini ya.

Membuat Models

Setelah kamu sudah membuat schemas yang diperlukan untuk operasi I/O data, selanjutnya kita akan beralih ke models. Models ini berisikan atribut-atribut yang ada didatabase, setelah itu baru bisa melakukan transaksi data dengan database. Selain itu juga models disini digunakan untuk ORM(Object Relational Model) yang dimana kita tidak menulis raw query sql melainkan hanya dengan menulis functionnya saja. Untuk lebih jelasnya nanti saya akan bahas pada routes.

Kita akan membuat folder baru bernama models lalu didalamnya ada file pegawai.py. Masukkan kode dibawah ini pada file models/pegawai.py

# models/pegawai.py
from sqlalchemy import (
    Column,
    Integer,
    MetaData,
    String,
    Table,
)
from sqlalchemy.sql.sqltypes import Date

metadata = MetaData()
Pegawai = Table(
    "dt_pegawai", metadata,
    Column("id_pegawai", Integer, primary_key=True, index=True),
    Column("nama_pegawai", String(255), nullable=False),
    Column("alamat_pegawai", String(255)),
    Column("ttl_pegawai", Date()),
    Column("telp_pegawai", String(255)),
    Column("email_pegawai", String(255)),
)

Penjelasan: Pada kode diatas terdapat 2 variable, yaitu metadata dan Pegawai. Pada variable metadata valuenya ialah function MetaData() dari sqlalchemy. Sedangkan pada variable Pegawai ialah function Table yang berfungsi untuk membuat table yang akan dijadikan orm nantinya. Didalam function Table pada params pertama ialah nama database dt_pegawai sedangkan pada params kedua ialah variable metadata yang tadi. Selanjutnya merupakan params Columns yang dimana kita akan menyesuaikan pada atribut-atribut sebelumnya yang kita buat di alembic tadi. Saya rasa kamu sudah sedikit memahami apa arti dari kode diatas.

Membuat Routes dan logic aplikasi

Pada step ini yang kita tunggu-tunggu. Pada step ini akan menjelaskan bagaimana proses terjadinya CRUD. Sebelum itu kita akan membuat folder baru bernama routes lalu didalamnya ada file pegawai.py. Masukkan kode dibawah ini pada file routes/pegawai.py

# routes/pegawai.py
from schemas.pegawai import PegawaiSchema, Pegawais
from models.pegawai import Pegawai
from fastapi import APIRouter, Response, status
from config.database import conn

pegawai = APIRouter() 

@pegawai.get('/pegawai/all', response_model=Pegawais, 
             description="Menampilkan semua data")
async def find_all_pegawai(limit: int = 10, offset: int = 0):
    query = Pegawai.select().offset(offset).limit(limit)
    data = conn.execute(query).fetchall()
    response = {"limit": limit, "offset": offset, "data": data }
    return response

@pegawai.get('/pegawai/{id}',
             description="Menampilkan detail data")
async def find_pegawai(id: int, response: Response):
    query = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    """
    Kenapa pakai huruf c pada Pegawai.c ?
    karena memakai ImmutableColumnCollection
    untuk melihat apa isinya silahkan uncomment print dibawah ini 
    dan lihat di terminal/cmd
    """
    #print(Pegawai.c)
    data = conn.execute(query).fetchone()
    if data is None:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"message": "data tidak ditemukan", "status": response.status_code}

    response = {"message": f"sukses mengambil data dengan id {id}", "data": data }
    return response

@pegawai.post('/pegawai/',
              description="Menambah data pegawai")
async def insert_pegawai(pgw : PegawaiSchema, response: Response):

    cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai)
    cek_email = conn.execute(cek_email).fetchone()
    if cek_email is not None:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {"status": response.status_code, "message": "email sudah digunakan"}

    query = Pegawai.insert().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    )
    # print(query)
    conn.execute(query)
    data = Pegawai.select().order_by(Pegawai.c.id_pegawai.desc())
    response = {"message": f"sukses menambahkan data baru", "data": conn.execute(data).fetchone() }
    return response

@pegawai.post('/pegawai/{id}',
              description="mengubah data pegawai")
async def update_pegawai(id: int, pgw : PegawaiSchema, response: Response):

    cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai, Pegawai.c.id_pegawai != id)
    cek_email = conn.execute(cek_email).fetchone()
    if cek_email is not None:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {"status": response.status_code, "message": "email sudah digunakan"}

    query = Pegawai.update().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    ).where(Pegawai.c.id_pegawai == id)
    # print(query)
    conn.execute(query)
    data = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    response = {"message": f"sukses mengubah data dengan id {id}", "data": conn.execute(data).fetchone() }
    return response

@pegawai.delete('/pegawai/{id}',
              description="menghapus data pegawai")
async def hapus_pegawai(id: int, response: Response):

    query = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    data = conn.execute(query).fetchone()
    if data is None:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"message": "data tidak ditemukan", "status": response.status_code}

    query = Pegawai.delete().where(Pegawai.c.id_pegawai == id)
    conn.execute(query)
    response = {"message": f"sukses menghapus data dengan id {id}" }
    return response

Penjelasan: Sebelum masuk ketiap function kita membuat variable bernama pegawai yang isinya function APIRouter() diimport dari fastapi. Variable pegawai ini nantinya sebagai @property decorator yang berisikan route aplikasi.

Proses menampilkan semua data

@pegawai.get('/pegawai/all', response_model=Pegawais, 
             description="Menampilkan semua data")
async def find_all_pegawai(limit: int = 10, offset: int = 0):
    query = Pegawai.select().offset(offset).limit(limit)
    data = conn.execute(query).fetchall()
    response = {"limit": limit, "offset": offset, "data": data }
    return response

Pada proses ini kita akan menampilkan semua data namun dengan batasan offset dan limit pada sql. Dengan begitu bisa diproses untuk pagination. Pertama kita mendefinisikan property decorator pegawai dengan object get(). Dalam get() berisikan params pertama yaitu route, param kedua yaitu response_model atau bisa dibilang sebagai pondai struktur pada json, param ketiga yaitu deskripsi dari route/method ini. Setelah itu function async find_all_pegawai() dengan params pertama limit tipe data integer default value 10, params kedua offset tipe data integer default value 0. Didalamnya terdapat 3 variable. Variable query yang berisikan value ORM query. jadi gk perlu nulis raw sql lagi. Cukup dengan menulis objectnya lalu generate ke raw sql. jika kalian mencoba print(query) maka akan menampilkan raw query sqlnya. Variable data ini berisikan value hasil eksekusi query ke database. Variable response ini kita membuat dictionary sebagai object sebelum menjadikannya json. Pastinya harus sesuai dengan schemas Pegawais ya. karena pada decorator kita diharuskan object sesuai seperti schemas. Setelah itu baru kita return dan akan menampilkan hasilnya.

Proses menampilkan detail data

@pegawai.get('/pegawai/{id}',
             description="Menampilkan detail data")
async def find_pegawai(id: int, response: Response):
    query = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    """
    Kenapa pakai huruf c pada Pegawai.c ?
    karena memakai ImmutableColumnCollection
    untuk melihat apa isinya silahkan uncomment print dibawah ini 
    dan lihat di terminal/cmd
    """
    #print(Pegawai.c)
    data = conn.execute(query).fetchone()
    if data is None:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"message": "data tidak ditemukan", "status": response.status_code}

    response = {"message": f"sukses mengambil data dengan id {id}", "data": data }
    return response

Pada proses ini kita akan menampilkan detail data berdasarkan id. Pertama kita mendefinisikan property decorator pegawai dengan object get(). Dalam get() berisikan params pertama yaitu route beserta id, param kedua yaitu deskripsi dari route/method ini. Setelah itu function async find_pegawai() dengan params pertama id tipe data integer, params kedua yaitu response dengan default value Response dari fastapi. Didalamnya terdapat 3 variable. Variable query yang berisikan value ORM query. jadi gk perlu nulis raw sql lagi. Cukup dengan menulis objectnya lalu generate ke raw sql. jika kalian mencoba print(query) maka akan menampilkan raw query sqlnya. Variable data ini berisikan value hasil eksekusi query ke database.

if data is None:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"message": "data tidak ditemukan", "status": response.status_code}

Sebelum ke variable response kita cek apakah hasil dari eksekusi query tidak kosong. Jika kosong maka aplikasi akan memberikan http status 404 karena tidak ada data. Variable response ini kita membuat dictionary sebagai object sebelum menjadikannya json. Setelah itu baru kita return dan akan menampilkan hasilnya.

Proses menambahkan data

@pegawai.post('/pegawai/',
              description="Menambah data pegawai")
async def insert_pegawai(pgw : PegawaiSchema, response: Response):

    cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai)
    cek_email = conn.execute(cek_email).fetchone()
    if cek_email is not None:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {"status": response.status_code, "message": "email sudah digunakan"}

    query = Pegawai.insert().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    )
    # print(query)
    conn.execute(query)
    data = Pegawai.select().order_by(Pegawai.c.id_pegawai.desc())
    response = {"message": f"sukses menambahkan data baru", "data": conn.execute(data).fetchone() }
    return response

Pada proses ini kita akan menambahkan data. Pertama kita mendefinisikan property decorator pegawai dengan object post(). Dalam post() berisikan params pertama yaitu route, param kedua yaitu deskripsi dari route/method ini. Setelah itu function async insert_pegawai() dengan params pertama pgw dengan default value PegawaiSchema yang diimport dari schemas/pegawai, params kedua yaitu response dengan default value Response dari fastapi. Sebelum melakukan tambah data, kita harus mengecek beberapa unique atribut.

cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai)
cek_email = conn.execute(cek_email).fetchone()
if cek_email is not None:
    response.status_code = status.HTTP_400_BAD_REQUEST
    return {"status": response.status_code, "message": "email sudah digunakan"}

Dalam project ini unique atribut ada pada atribut email. Sehingga client menambahkan data email yang sudah ada maka akan memberi respon 400 / bad request, yang dikarenakan data sudah ada sebelumnya. ketika client telah lolos seleksi cek email, lalu akan menerima inputan client berdasarkan sesuai dengan schemas dengan variable pgw. Setelah itu disimpan pada variable query untuk membentuk raw query berdasarkan ORM yang kita buat.

query = Pegawai.insert().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    )
# print(query)
conn.execute(query)
data = Pegawai.select().order_by(Pegawai.c.id_pegawai.desc())

Setelah query telah terbuat, lalu sistem akan mengeksekusi query tersebut dengan con.execute(). Otomatis data akan sudah ada di database. Setelah itu kita akan mengambil data yang baru tersebut dengan ORM lalu simpan ke variable data. Pada ORM data kita hanya mengambil data yang terbaru yang barusan kita input. Ini opsional ya. jika kalian tidak ingin menampilkan datanya ya silahkan saja. Setelah itu kita tinggal membuat variable response sebagai wadah dari output lalu return dengan variable response tersebut.

response = {"message": f"sukses menambahkan data baru", "data": conn.execute(data).fetchone() }
return response

 

Proses mengubah data

@pegawai.post('/pegawai/{id}',
              description="mengubah data pegawai")
async def update_pegawai(id: int, pgw : PegawaiSchema, response: Response):

    cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai, Pegawai.c.id_pegawai != id)
    cek_email = conn.execute(cek_email).fetchone()
    if cek_email is not None:
        response.status_code = status.HTTP_400_BAD_REQUEST
        return {"status": response.status_code, "message": "email sudah digunakan"}

    query = Pegawai.update().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    ).where(Pegawai.c.id_pegawai == id)
    # print(query)
    conn.execute(query)
    data = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    response = {"message": f"sukses mengubah data dengan id {id}", "data": conn.execute(data).fetchone() }
    return response

Pada proses ini kita akan mengubah data. Pertama kita mendefinisikan property decorator pegawai dengan object post(). Dalam post() berisikan params pertama yaitu route beserta id, param kedua yaitu deskripsi dari route/method ini. Setelah itu function async update_pegawai() dengan params pertama id yang bertipe data integer, params kedua pgw dengan default value PegawaiSchema yang diimport dari schemas/pegawai, params ketiga yaitu response dengan default value Response dari fastapi. Sebelum melakukan update data, kita harus mengecek beberapa unique atribut.

cek_email = Pegawai.select().filter(Pegawai.c.email_pegawai == pgw.email_pegawai, Pegawai.c.id_pegawai != id)
cek_email = conn.execute(cek_email).fetchone()
if cek_email is not None:
    response.status_code = status.HTTP_400_BAD_REQUEST
    return {"status": response.status_code, "message": "email sudah digunakan"}

Dalam project ini unique atribut ada pada atribut email. Email ini dikhususkan bukan id yang sama, dalam arti query akan mengecek hanya yang bukan idnya untuk email yang sama. Sehingga client menambahkan data email yang sudah ada maka akan memberi respon 400 / bad request, yang dikarenakan data sudah ada sebelumnya. ketika client telah lolos seleksi cek email, lalu akan menerima inputan client berdasarkan sesuai dengan schemas dengan variable pgw. Setelah itu disimpan pada variable query untuk membentuk raw query berdasarkan ORM yang kita buat.

query = Pegawai.update().values(
        nama_pegawai = pgw.nama_pegawai,
        alamat_pegawai = pgw.alamat_pegawai,
        ttl_pegawai = pgw.ttl_pegawai,
        telp_pegawai = pgw.telp_pegawai,
        email_pegawai = pgw.email_pegawai
    ).where(Pegawai.c.id_pegawai == id)
# print(query)
conn.execute(query)
data = Pegawai.select().where(Pegawai.c.id_pegawai == id)

Setelah query telah terbuat, lalu sistem akan mengeksekusi query tersebut dengan con.execute(). Otomatis data akan sudah terupdate di database. Setelah itu kita akan mengambil data yang baru tersebut dengan ORM lalu simpan ke variable data. Pada ORM data kita hanya mengambil data berdasarkan id yang kita pilih di route. Ini opsional ya. jika kalian tidak ingin menampilkan datanya ya silahkan saja. Setelah itu kita tinggal membuat variable response sebagai wadah dari output lalu return dengan variable response tersebut.

response = {"message": f"sukses mengubah data dengan id {id}", "data": conn.execute(data).fetchone() }
return response

 

Proses Menghapus Data

@pegawai.delete('/pegawai/{id}',
              description="menghapus data pegawai")
async def hapus_pegawai(id: int, response: Response):
    query = Pegawai.select().where(Pegawai.c.id_pegawai == id)
    data = conn.execute(query).fetchone()
    if data is None:
        response.status_code = status.HTTP_404_NOT_FOUND
        return {"message": "data tidak ditemukan", "status": response.status_code}

    query = Pegawai.delete().where(Pegawai.c.id_pegawai == id)
    conn.execute(query)
    response = {"message": f"sukses menghapus data dengan id {id}" }
    return response

Pada proses ini kita akan menampilkan menghapus data berdasarkan id. Pertama kita mendefinisikan property decorator pegawai dengan object delete(). Dalam delete() berisikan params pertama yaitu route beserta id, param kedua yaitu deskripsi dari route/method ini. Setelah itu function async hapus_pegawai() dengan params pertama id tipe data integer, params kedua yaitu response dengan default value Response dari fastapi. langkah pertama kita akan mencari data dengan id tersebut. Jika data ditemukan maka akan ke proses selanjutnya. Jika data tidak ditemukan maka dia akan memberikan response 404 atau data tidak ada.

query = Pegawai.select().where(Pegawai.c.id_pegawai == id)
data = conn.execute(query).fetchone()
if data is None:
    response.status_code = status.HTTP_404_NOT_FOUND
    return {"message": "data tidak ditemukan", "status": response.status_code}

Setelah itu kita akan menghapusnya berdasarkan id. lalu kita return berupa sukses telah menghapus data.

query = Pegawai.delete().where(Pegawai.c.id_pegawai == id)
conn.execute(query)
response = {"message": f"sukses menghapus data dengan id {id}" }
return response

Buat file main

Pada langkah terakhir ini kita sudha berada pada tahap terakhir, buat file main.py pada root project.

#/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routes.pegawai import pegawai

app = FastAPI()

def cors_headers(app):
    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_methods=["*"],
        allow_headers=["*"],
        allow_credentials=True,
        )
    return app

app.include_router(pegawai)

@app.get("/")
async def root():
    return {"message": "CRUD FastApi Mysql","Authors":"Afrizals Blog"}

Penjelasan: Pertama kita membutuhkan 3 module, yaitu

  • FastAPI sebagai insisialisasi aplikasi
  • CORSMiddleware sebagai konfigurasi CORS Headers
  • Routes Pegawai sebagai semua routes yang ada pada pegawai

Lalu kita inisilisasikan variable app sebagai FastAPI(). Kita akan buat function cors_headers sebagai konfigurasi CORS. Silahkan baca dokumentasi ini untuk lebih lengkapnya tentang CORS pada FastAPI. Setelah itu kita panggil method include_router() sebagai inisialisasi bahwa kita juga memanggil routes pegawai app.include_router(pegawai) Terakhir pastinya kita membuat route root atau / dengan function dibawahnya saya beri nama root().

Sehingga struktur direktori akhir akan seperti ini

├── alembic
│   ├── env.py
│   ├── README
│   ├── script.py.mako
│   └── versions
│       └── a94e7a90e9fa_create_table_pegawai.py
├── alembic.ini
├── config
│   └── database.py
├── main.py
├── models
│   └── pegawai.py
├── requirements.txt
├── routes
│   └── pegawai.py
└── schemas
    └── pegawai.py

Jalankan server

Setelah semua telah selesai silahkan untuk menjalankan servernya menggunakan uvicorn dengan perintah dibawah ini

uvicorn main:app --reload

Lalu buka url ke alamat http://127.0.0.1:8000/docs sebagai ujicoba CRUDnya

Testing (bonus)

Pada subbab ini saya berikan testing sederhana menggunakan pytest. Silahkan copy kode dibawah ini dan pastekan kedalam file baru bernama test_main.py

# /test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "CRUD FastApi Mysql","Authors":"Afrizals Blog"}

def test_crud_user():
    # create data
    response = client.post(
        "/pegawai/",
        json={
                "nama_pegawai": "budi",
                "alamat_pegawai": "surabaya",
                "ttl_pegawai": "1997-10-03",
                "telp_pegawai": "4045",
                "email_pegawai": "budi@domain.com"
            },
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data['data']['email_pegawai'] == "budi@domain.com"
    assert "id_pegawai" in data['data']
    user_id = data['data']['id_pegawai']

    # read detail data
    response = client.get(f"/pegawai/{user_id}")
    assert response.status_code == 200, response.text
    data = response.json()
    assert data['data']['email_pegawai'] == "budi@domain.com"
    assert data['data']['id_pegawai'] == user_id

    # update data
    response = client.post(
        f"/pegawai/{user_id}",
        json={
                "nama_pegawai": "budi update",
                "alamat_pegawai": "surabaya",
                "ttl_pegawai": "1997-10-03",
                "telp_pegawai": "4045",
                "email_pegawai": "budi_update@domain.com"
            },
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data['data']['nama_pegawai'] == "budi update"
    assert data['data']['email_pegawai'] == "budi_update@domain.com"

    # delete data
    response = client.delete(
        f"/pegawai/{user_id}"
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data['message'] == f"sukses menghapus data dengan id {user_id}"

Pastikan hasilnya 2 passed ya

============================= test session starts ==============================
platform linux -- Python 3.9.5, pytest-6.2.5, py-1.10.0, pluggy-1.0.0
rootdir: /project/kamu/fastapi-mysql
plugins: Faker-8.14.0
collected 2 items                                                              

test_main.py ..                                                          [100%]

============================== 2 passed in 0.74s ===============================

Kesimpulan

Pada tutorial ini kita sudah belajar bagaimana caranya membuat Migration,Seed,dan CRUD menggunakan FastAPI dengan database MySQL. Cukup sekian dari tutorial ini, semoga bermanfaat. Silahkan tunggu konten lainnya tentang FastAPI di blog ini.

Apabila artikel ini membantu kamu, dan jika kamu memiliki sedikit rezeki, mohon dukungannya dalam bentuk segelas kopi melalui URL dibawah ini:

Tags:
#Python #FastAPI #MySql
Last Updated: Minggu, 3 Oktober 2021 13:42 GMT+7

Comments

Related Posts

© 2026 Afrizal Muhammad Yasin. Made with ❤️

For business / recruiting: me@afrizalmy.com