6 min read

Dimulai dari 0

Dimulai dari 0
Photo by David Vilches / Unsplash

Selama ini saya bekerja berpindah-pindah stack, awal menggunakan framework Codeignter dan Laravel, pindah ke python langsung menggunakan Django dan sekarang di kantor menggunakan Go yang sudah dibuat struktur dan pattern oleh atasan saya, karena hal itu saya berpikir saya sangat jarang bikin sesuatu yang emang dari awal, saya hanyalah penambah fitur saja di atas framework yang sudah jadi.

Dari pengalaman di atas, di waktu libur natal dan tahun baru ini kepikiran mari buat sesuatu dari "awal" tentu gak dari awal banget, saya masih pakai kode orang lain (fast api) tapi mari coba untuk bikin struktur yang memang pilihan personal baik minus dan plusnya.

Memilih bahasa.

Awalnya saya ingin membuat kerangka kerja ini dengan Go dengan alasan sekalian belajar lebih tentang Go, tapi untuk sekarang saya butuh hasil yang cepat untuk memastikan POC struktur saya cukup oke, maka dari itu saya menggunakan yang lebih familiar yaitu python dengan FastApi.

Struktur dan Inspirasi

Inspirasi saya adalah Django, saya suka bagaimana django mengunakan istilah "app", dan di dalam "app" tersebut berisi hal-hal untuk menghasilkan 1 buah projek, secara kasar sebenarnya kita bisa buat seluruh aplikasi django kita dalam satu "app" kan? Oke cukup tentang django, lalu bagaimana dengan struktur yang saya mau, beginilah struktur kasar yang saya bayangkan.

project/
├── main.py # FastAPI application entry point
├── app/
│ └── user/
│ ├── routes.py # API routes
│ ├── service.py # Business logic
│ ├── storage.py # Database operations
├── conf/
│ ├── database.py 
│ └── logger.py
└── sql/
└── schema.sql # Database schema
└── tests/
│ └── user/
│ ├── storage_test.py 
└── conftest.sql

main.py

Berkas ini adalah pintu masuk aplikasi, merajut segala konfigurasi sampai aplikasi bisa dijalankan dengan baik dan digunakan. Tidak ada yang spesial di sini.

app

Ini dalah folder mengikuti pola django dengan istilah app, sayapun menggunakan pendekatan yang sama, bedanya jika app django di-level root maka saya dibungkus semua di dalam folder app

"Module"

user
ini adalah folder user atau app user, di sini semua hal yang berkaitan dengan user akan disimpan di sini, nantinya jika punya fitur lain seperti order maka saya akan buat folder order di dalam app.

Mungkin beberapa akan mengira apakah ini DDD? Saya tidak bisa bilang ini DDD karena pemahaman saya yang belum terlalu baik dengan apa itu DDD maka menyebut pola ini DDD hanya akan berujung kesalah pahaman, yang pasti dari benak saya setiap "app" adalah "independent" jika antar app akan saling berkomunikasi maka pola yang digunakan adalah bayangkan saja "app" yang lain itu adalah "third party"

CONF

Ini adalah folder berisi konfigurasi, saat ini hanya ada database dan logger. Nantinya untuk hal-hal bersifat konfigurasi apapun akan disimpan di sini dan akan dirajut di main.py

SQL

Saya memilih untuk tidak menggunakan ORM sehingga informasi migration dan hal-hal terkait sql akan ditampung di sini.

Tests

Sesuai namanya folder ini bertanggung jawab tentang segala macam test.

Setelah membahas masing-masing folder sekarang bahas apa saja berkas di dalam module app ambil contoh adalah module user.

data.py: Berkas ini bertanggung jawab untuk informasi segala macam urusan mengenai data dari module user, mungkin bsia dibilang seperti DTO, tapi contoh data.py yang saya buat untuk projek ini adalah seperti berikut

# This data.py file acts as a DTO (Data Transfer Object). 
# The information here will be passed around to other applications. 
# The dataclasses defined can either directly match the database schema 
# or differ, depending on the requirements.


from pydantic import BaseModel
from common.public_struct import  Request, Response
from typing import Optional

# User data structure for user
class User(BaseModel):
    id: Optional[int] = None
    name: str
    username: str
    email: str
    password: str | None


# Data input output for service
class GetUserInput(BaseModel):
    trace_id : str
    user_id : int

class GetUserOutput(BaseModel):
    success: bool = False
    msg: Optional[str] = ""
    # for testing
    user: Optional[User] = None

class CreateUserInput(BaseModel):
    trace_id : str
    name : Optional[str] = None
    email: str
    username: str
    password: str

class CreateUserOutput(BaseModel):
    success: bool = False
    msg: Optional[str] = ""
    # for testing
    Id: Optional[int] = 0

class LoginInput(BaseModel):
    trace_id : str
    email: str
    password: str

class LoginOutput(BaseModel):
    success: bool = False
    msg: Optional[str] = ""


# Data for request Response
class UserDetailResp(Response):
    data: Optional[User] = None

class CreateUserReq(Request):
    email: str
    username: str
    password: str

class CreateUserResp(Response):
    pass

class LoginReq(Request):
    email: str
    password: str

class LoginResp(Response):
    pass

storage.py: Berkas ini berfungsi sebagai penghubung antara aplikasi dan database, karena saya tidak menggunakan ORM mau tidak mau harus menulis manual.

from app.user.data import User

async def insert_user(connection, user_data) -> int:
    query = """
        INSERT INTO users (fullname, username, email, password_hash)
        VALUES ($1, $2, $3, $4)
        RETURNING id
    """
    result = await connection.fetchval(
        query,
        user_data.name,
        user_data.username,
        user_data.email,
        user_data.password,
    )
    return result  # Only return the `id`

async def get_user_by_id(connection, user_id: int) -> User | None:
    query = "SELECT id, fullname, username, email, password_hash FROM users WHERE id = $1"
    record = await connection.fetchrow(query, user_id)
    if record:
        return User(
            id=record['id'],
            name=record['fullname'],
            email=record['email'],
            username=record['username'],
            password=record['password_hash']
        )
    return None

service.py: Tempat saya menaruh fungsi utama dari aplikasi, dalam asumsi saya ini bersifat "agnostik" bisa dipanggil nantinya dari endpoint api atau sekadar cron.

Contoh service.py

async def create_user(user_input: CreateUserInput) -> CreateUserOutput:
    resp = CreateUserOutput()
    if user_input.password == "":
        resp.msg = "Password wajib diisi."
        return  resp

    if user_input.username == "":
        resp.msg = "Username wajib diisi."
        return resp

    if user_input.email == "":
        resp.msg = "Email wajib diisi."
        return resp

    if len(user_input.username) < 5:
        resp.msg = "username minimal 5 karakter."
        return resp

    if len(user_input.email) < 5:
        resp.msg = "email minimal 5 karakter."
        return resp

    if len(user_input.password) < 5:
        resp.msg = "pasword minimal 5 karakter."
        return resp

    passwords = create_django_like_pbkdf2_password(user_input.password)
    user = User(
        username=user_input.username,
        email=user_input.email,
        password=passwords,
        name=user_input.name,
    )
    async with db.pool.acquire() as connection:
        tx = connection.transaction()  # Begin a new transaction
        logger.info(trace_id=user_input.trace_id, msg="Begin tx")
        await tx.start()

        try:
            # Insert user and get the new ID
            new_user_id = await insert_user(connection, user)
            # Commit transaction if everything is successful
            await tx.commit()
            resp.Id = new_user_id
            resp.success = True
            return resp
        except Exception as e:
            # Rollback transaction if something goes wrong
            logger.err(trace_id=user_input.trace_id, msg="Failed to commit")
            await tx.rollback()
            resp.msg = "terjadi kesalahan dalam sistem"
            return resp

routes.py: Tugas berkas ini sederhana, tidak ada logic yang "berat" karena itu tugasnya service, tugas berkas ini cukup menerima request, serahkan ke service, dan kirimkan response.

Progress

Saat ini progress yang berjalan adalah aplikasi berjalan sudah sesuai yang diharapkan, proses yang sedikit lama adalah bagaimana menyusun test parallel dengan multiple db, prasyarat test yang saya mau:

  1. Tes berjalan parallel
  2. Setiap test akan diisiolasi dengan database sendiri sehingga akan lebih mudah untuk bikin test dengan data "asli".
  3. Setiap test dijalankan akan melakukan koneksi ke salah satu database testing, melakukan lock, execute schema dari folder sql, menjalankan test, bersihkan database, release lock sehingga ketika ada test yang sedang mengantri bisa menggunakan db yang tadi tapi dalam kondisi bersih.

Berikut hasil konfigurasi yang sudah saya buat

Inspirasi:

Django
The web framework for perfectionists with deadlines.
Django Project
Running Flask Tests In Parallell with Pytest
Running tests in parallel with pytest
IT projects under active development tend to grow a lot. So their test suites. It is only a matter of time until you’ll be looking for ways to speed up the execution of tests.
breadcrumbs collector.tech

Selain tiga web ini tentu saja "rekan pair programming" claude sone dan gpt-4o (keduanya versi gratis)

Published by:

aries