Membuat Aplikasi Web CRUD Sederhana dengan Flask dan MySQL
Pengenalan: Flask untuk Pemula
Halo teman-teman developer! Di artikel kali ini, kita akan belajar bersama cara membuat aplikasi web CRUD (Create, Read, Update, Delete) sederhana menggunakan framework Flask dan database MySQL. Flask adalah salah satu framework web Python yang paling populer karena kesederhanaannya namun tetap powerful.
Kita akan mulai dari nol: setup environment, install dependencies, buat database, implementasi CRUD operations, sampai dengan fitur-fitur advanced seperti authentication, testing, dan deployment. Cocok banget buat kamu yang baru mulai belajar Flask atau ingin refresh pengetahuan tentang web development dengan Python.
Daftar Isi
- Mengapa Flask + MySQL?
- Persiapan Environment
- Setup Database MySQL
- Implementasi CRUD Dasar
- Struktur Folder yang Baik
- Authentication dan Authorization
- Error Handling dan Validation
- Testing dengan pytest
- Caching untuk Performance
- Background Tasks dengan Celery
- API Documentation dengan Swagger
- Deployment dengan Gunicorn
- Kesimpulan
Mengapa Flask + MySQL?
Flask adalah micro-framework web Python yang ringan namun extensible. Berbeda dengan Django yang “batteries included”, Flask memberi kamu kebebasan penuh untuk memilih tools yang kamu butuhkan. Ini membuat Flask sangat cocok untuk:
- Microservices architecture: Ringan dan mudah di-scale
- API development: Mudah diintegrasikan dengan frontend modern
- Prototyping: Cepat develop tanpa boilerplate code
- Learning purposes: Kurva belajar yang gentle untuk pemula
MySQL sebagai database pilihan karena:
- Reliable: Sudah terbukti di production environment
- Fast: Query performance yang excellent
- Widely used: Banyak tools dan community support
- ACID compliant: Data integrity terjamin
Persiapan Environment
Install Python
Pastikan Python 3.8+ sudah terinstall di sistem kamu. Cek dengan:
python --version
# atau
python3 --version
Jika belum ada, download dari python.org atau gunakan package manager sistem kamu.
Virtual Environment
Selalu gunakan virtual environment untuk setiap project Python. Ini mencegah konflik dependency antar project.
# Buat folder project
mkdir flask-crud-app
cd flask-crud-app
# Buat virtual environment
python -m venv venv
# Aktifkan virtual environment
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate
Install Dependencies
Install Flask dan library pendukung:
pip install Flask flask-sqlalchemy pymysql python-dotenv pytest flask-testing
Kita akan install library tambahan seiring development berjalan. Untuk SQLAlchemy dengan MySQL, kita gunakan PyMySQL sebagai driver.
Setup Database MySQL
Install MySQL Server
# Ubuntu/Debian
sudo apt update
sudo apt install mysql-server
# CentOS/RHEL
sudo yum install mysql-server
# macOS dengan Homebrew
brew install mysql
Start MySQL service dan setup:
sudo systemctl start mysql # Linux
sudo systemctl enable mysql
# Setup secure installation
sudo mysql_secure_installation
Buat Database dan Table
Login ke MySQL sebagai root:
mysql -u root -p
Buat database dan table untuk contoh CRUD kita:
-- Buat database
CREATE DATABASE flask_crud_app;
-- Gunakan database
USE flask_crud_app;
-- Buat table users (untuk authentication nanti)
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
password_hash VARCHAR(255) NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);
-- Buat table posts (untuk CRUD operations)
CREATE TABLE posts (
id INT AUTO_INCREMENT PRIMARY KEY,
title VARCHAR(200) NOT NULL,
content TEXT,
author_id INT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
FOREIGN KEY (author_id) REFERENCES users(id) ON DELETE CASCADE
);
-- Insert sample data
INSERT INTO users (username, email, password_hash) VALUES
('admin', '[email protected]', 'hashed_password_here'),
('user1', '[email protected]', 'hashed_password_here');
Implementasi CRUD Dasar
Struktur Aplikasi
Buat struktur folder seperti ini:
flask-crud-app/
├── app/
│ ├── __init__.py
│ ├── models.py
│ ├── routes.py
│ ├── auth.py
│ ├── forms.py
│ └── templates/
│ ├── base.html
│ ├── index.html
│ ├── login.html
│ └── post_form.html
├── config.py
├── requirements.txt
└── run.py
File Konfigurasi
Buat config.py:
import os
from dotenv import load_dotenv
load_dotenv()
class Config:
SECRET_KEY = os.getenv('SECRET_KEY', 'dev-secret-key')
# SQLAlchemy configuration
SQLALCHEMY_DATABASE_URI = os.getenv(
'DATABASE_URL',
'mysql+pymysql://root:@localhost/flask_crud_app'
)
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = False # Set True untuk debug SQL queries
# Pagination
POSTS_PER_PAGE = 10
# Upload settings
UPLOAD_FOLDER = 'app/static/uploads'
MAX_CONTENT_LENGTH = 16 * 1024 * 1024 # 16MB
class DevelopmentConfig(Config):
DEBUG = True
class ProductionConfig(Config):
DEBUG = False
config = {
'development': DevelopmentConfig,
'production': ProductionConfig,
'default': DevelopmentConfig
}
Models
Buat app/models.py:
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash
from datetime import datetime
db = SQLAlchemy()
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False)
email = db.Column(db.String(100), unique=True, nullable=False)
password_hash = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Relationship
posts = db.relationship('Post', backref='author', lazy='dynamic', cascade='all, delete-orphan')
def __repr__(self):
return f'<User {self.username}>'
@staticmethod
def get_by_id(user_id):
return User.query.get(user_id)
@staticmethod
def get_by_username(username):
return User.query.filter_by(username=username).first()
@staticmethod
def get_by_email(email):
return User.query.filter_by(email=email).first()
def set_password(self, password):
self.password_hash = generate_password_hash(password)
def check_password(self, password):
return check_password_hash(self.password_hash, password)
def save(self):
db.session.add(self)
db.session.commit()
def delete(self):
db.session.delete(self)
db.session.commit()
class Post(db.Model):
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False)
content = db.Column(db.Text)
author_id = db.Column(db.Integer, db.ForeignKey('users.id'), nullable=False)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f'<Post {self.title}>'
@staticmethod
def get_all(page=1, per_page=10):
return Post.query.order_by(Post.created_at.desc()).paginate(
page=page, per_page=per_page, error_out=False
)
@staticmethod
def get_by_id(post_id):
return Post.query.get(post_id)
@staticmethod
def get_by_author(author_id, page=1, per_page=10):
return Post.query.filter_by(author_id=author_id).order_by(
Post.created_at.desc()
).paginate(page=page, per_page=per_page, error_out=False)
@staticmethod
def search(keyword, page=1, per_page=10):
return Post.query.filter(
(Post.title.contains(keyword)) | (Post.content.contains(keyword))
).order_by(Post.created_at.desc()).paginate(
page=page, per_page=per_page, error_out=False
)
def save(self):
db.session.add(self)
db.session.commit()
def delete(self):
db.session.delete(self)
db.session.commit()
Keuntungan menggunakan SQLAlchemy:
- Type-safe: Column types didefinisikan dengan jelas
- Relationships: Foreign key dan relationships sudah built-in
- Query builder: Lebih readable dan maintainable
- Migration support: Mudah untuk database schema changes
- Protection: Built-in SQL injection protection
- Pagination: Built-in pagination support
Initialize Database
Buat app/__init__.py:
from flask import Flask
from config import config
from .models import db
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
# Register blueprints
from .routes import main
from .auth import auth
app.register_blueprint(main)
app.register_blueprint(auth)
# Create tables
with app.app_context():
db.create_all()
return app
Routes CRUD
Buat app/routes.py:
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
from .models import Post
from .forms import PostForm
from functools import wraps
import math
main = Blueprint('main', __name__)
def login_required(f):
@wraps(f)
def decorated_function(*args, **kwargs):
if 'user_id' not in session:
return redirect(url_for('auth.login'))
return f(*args, **kwargs)
return decorated_function
@main.route('/')
def index():
page = request.args.get('page', 1, type=int)
pagination = Post.get_all(page=page)
posts = pagination.items
return render_template('index.html',
posts=posts,
pagination=pagination)
@main.route('/post/new', methods=['GET', 'POST'])
@login_required
def new_post():
form = PostForm()
if form.validate_on_submit():
post = Post(
title=form.title.data,
content=form.content.data,
author_id=session['user_id']
)
post.save()
flash('Post created successfully!', 'success')
return redirect(url_for('main.index'))
return render_template('post_form.html', form=form, title='New Post')
@main.route('/post/<int:post_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_post(post_id):
post = Post.get_by_id(post_id)
if not post:
flash('Post not found', 'error')
return redirect(url_for('main.index'))
# Check if user owns the post
if post.author_id != session['user_id']:
flash('You can only edit your own posts', 'error')
return redirect(url_for('main.index'))
form = PostForm(obj=post)
if form.validate_on_submit():
post.title = form.title.data
post.content = form.content.data
post.save()
flash('Post updated successfully!', 'success')
return redirect(url_for('main.index'))
return render_template('post_form.html', form=form, title='Edit Post')
@main.route('/post/<int:post_id>/delete', methods=['POST'])
@login_required
def delete_post(post_id):
post = Post.get_by_id(post_id)
if post and post.author_id == session['user_id']:
post.delete()
flash('Post deleted successfully!', 'success')
else:
flash('Post not found or you do not have permission', 'error')
return redirect(url_for('main.index'))
Struktur Folder yang Baik
Organisasi folder yang baik membantu maintainability aplikasi:
flask-crud-app/
├── app/
│ ├── api/ # API blueprints
│ ├── auth/ # Authentication logic
│ ├── models/ # Database models
│ ├── routes/ # Route handlers
│ ├── static/ # CSS, JS, images
│ ├── templates/ # Jinja2 templates
│ ├── utils/ # Helper functions
│ └── __init__.py
├── migrations/ # Database migrations
├── tests/ # Unit and integration tests
├── docs/ # Documentation
├── .env # Environment variables
├── config.py # Configuration
├── requirements.txt # Dependencies
└── run.py # Application entry point
Database Migration dengan Flask-Migrate
Ketika menggunakan SQLAlchemy, kamu perlu tool untuk manage perubahan schema database. Flask-Migrate (wrapper untuk Alembic) sangat berguna untuk ini.
Install Flask-Migrate
pip install Flask-Migrate
Setup Migration
Update app/__init__.py:
from flask import Flask
from flask_migrate import Migrate
from config import config
from .models import db
migrate = Migrate()
def create_app(config_name='default'):
app = Flask(__name__)
app.config.from_object(config[config_name])
# Initialize extensions
db.init_app(app)
migrate.init_app(app, db)
# Register blueprints
from .routes import main
from .auth import auth
app.register_blueprint(main)
app.register_blueprint(auth)
return app
Initialize Migration Repository
# Initialize migration folder
flask db init
# Create first migration
flask db migrate -m "Initial migration"
# Apply migration
flask db upgrade
Alur Kerja Migration
Setiap kali kamu ubah model:
# 1. Edit model di models.py
# 2. Generate migration script
flask db migrate -m "Add column email_verified to users"
# 3. Review migration file di migrations/versions/
# 4. Apply migration
flask db upgrade
Untuk rollback:
# Rollback satu versi
flask db downgrade
# Rollback ke versi tertentu
flask db downgrade <revision_id>
Tips Migration
- Selalu review migration script sebelum apply
- Backup database sebelum run migration di production
- Test migration di development environment dulu
- Commit migration files ke version control
Authentication dan Authorization
Implementasi login/logout dengan session:
# app/auth.py
from flask import Blueprint, render_template, request, redirect, url_for, flash, session
from .models import User
from .forms import LoginForm, RegisterForm
from werkzeug.security import generate_password_hash
auth = Blueprint('auth', __name__)
@auth.route('/login', methods=['GET', 'POST'])
def login():
form = LoginForm()
if form.validate_on_submit():
user = User.get_by_username(form.username.data)
if user and user.check_password(form.password.data):
session['user_id'] = user.id
session['username'] = user.username
flash('Login successful!', 'success')
return redirect(url_for('main.index'))
flash('Invalid username or password', 'error')
return render_template('login.html', form=form)
@auth.route('/register', methods=['GET', 'POST'])
def register():
form = RegisterForm()
if form.validate_on_submit():
if User.get_by_username(form.username.data):
flash('Username already exists', 'error')
return render_template('register.html', form=form)
user = User(
username=form.username.data,
email=form.email.data
)
user.set_password(form.password.data)
user.save()
flash('Registration successful! Please login.', 'success')
return redirect(url_for('auth.login'))
return render_template('register.html', form=form)
@auth.route('/logout')
def logout():
session.clear()
flash('You have been logged out', 'info')
return redirect(url_for('main.index'))
Error Handling dan Validation
Buat custom error handlers dan form validation:
# app/forms.py
from flask_wtf import FlaskForm
from wtforms import StringField, TextAreaField, PasswordField
from wtforms.validators import DataRequired, Length, Email, EqualTo
class LoginForm(FlaskForm):
username = StringField('Username', validators=[DataRequired()])
password = PasswordField('Password', validators=[DataRequired()])
class RegisterForm(FlaskForm):
username = StringField('Username', validators=[DataRequired(), Length(min=3, max=50)])
email = StringField('Email', validators=[DataRequired(), Email()])
password = PasswordField('Password', validators=[DataRequired(), Length(min=6)])
confirm_password = PasswordField('Confirm Password',
validators=[DataRequired(), EqualTo('password')])
class PostForm(FlaskForm):
title = StringField('Title', validators=[DataRequired(), Length(max=200)])
content = TextAreaField('Content', validators=[DataRequired()])
# Error handlers in __init__.py
@app.errorhandler(404)
def not_found(error):
return render_template('404.html'), 404
@app.errorhandler(500)
def internal_error(error):
db.session.rollback()
return render_template('500.html'), 500
Advanced SQLAlchemy Features
Database Constraints dan Indexes
Tambahkan constraints dan indexes untuk data integrity dan performance:
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(50), unique=True, nullable=False, index=True)
email = db.Column(db.String(100), unique=True, nullable=False, index=True)
password_hash = db.Column(db.String(255), nullable=False)
is_active = db.Column(db.Boolean, default=True)
created_at = db.Column(db.DateTime, default=datetime.utcnow)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
# Table args untuk composite indexes
__table_args__ = (
db.Index('idx_username_email', 'username', 'email'),
db.CheckConstraint('LENGTH(username) >= 3', name='username_min_length'),
)
class Post(db.Model):
__tablename__ = 'posts'
id = db.Column(db.Integer, primary_key=True)
title = db.Column(db.String(200), nullable=False, index=True)
slug = db.Column(db.String(250), unique=True, index=True)
content = db.Column(db.Text)
author_id = db.Column(db.Integer, db.ForeignKey('users.id', ondelete='CASCADE'))
status = db.Column(db.Enum('draft', 'published', 'archived', name='post_status'),
default='draft')
view_count = db.Column(db.Integer, default=0)
created_at = db.Column(db.DateTime, default=datetime.utcnow, index=True)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
__table_args__ = (
db.Index('idx_author_status', 'author_id', 'status'),
db.Index('idx_status_created', 'status', 'created_at'),
)
Query Optimization
Gunakan lazy loading dan eager loading sesuai kebutuhan:
# Lazy loading (default) - query terpisah
post = Post.query.get(1)
print(post.author.username) # Trigger query kedua
# Eager loading dengan joinedload - satu query dengan JOIN
from sqlalchemy.orm import joinedload
posts = Post.query.options(joinedload(Post.author)).all()
for post in posts:
print(post.author.username) # Tidak trigger query lagi
# Subquery loading - dua query tapi lebih efisien untuk one-to-many
from sqlalchemy.orm import subqueryload
users = User.query.options(subqueryload(User.posts)).all()
# Select specific columns untuk reduce memory
posts = db.session.query(Post.id, Post.title, Post.created_at).all()
Custom Query Methods
Tambahkan method helper untuk queries yang sering dipakai:
class Post(db.Model):
# ... existing code ...
@staticmethod
def get_published(page=1, per_page=10):
return Post.query.filter_by(status='published').order_by(
Post.created_at.desc()
).paginate(page=page, per_page=per_page, error_out=False)
@staticmethod
def get_popular(limit=10):
return Post.query.filter_by(status='published').order_by(
Post.view_count.desc()
).limit(limit).all()
@staticmethod
def get_by_tag(tag_name, page=1, per_page=10):
return Post.query.join(Post.tags).filter(
Tag.name == tag_name
).paginate(page=page, per_page=per_page, error_out=False)
def increment_view_count(self):
self.view_count += 1
db.session.commit()
Validation Mixins
Buat reusable validation patterns:
from sqlalchemy.orm import validates
class User(db.Model):
# ... existing code ...
@validates('email')
def validate_email(self, key, email):
if '@' not in email:
raise ValueError('Invalid email address')
return email.lower()
@validates('username')
def validate_username(self, key, username):
if len(username) < 3:
raise ValueError('Username must be at least 3 characters')
if not username.isalnum():
raise ValueError('Username must be alphanumeric')
return username
class Post(db.Model):
# ... existing code ...
@validates('title')
def validate_title(self, key, title):
if len(title) < 5:
raise ValueError('Title must be at least 5 characters')
return title
Testing dengan pytest
Buat file tests/test_app.py:
import pytest
from app import create_app
from app.models import db, User, Post
@pytest.fixture
def app():
app = create_app('testing')
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
with app.app_context():
db.create_all()
yield app
db.session.remove()
db.drop_all()
@pytest.fixture
def client(app):
return app.test_client()
def test_home_page(client):
response = client.get('/')
assert response.status_code == 200
def test_user_registration(app):
with app.app_context():
# Test user creation
user = User(username='testuser', email='[email protected]')
user.set_password('password123')
user.save()
# Retrieve and verify
retrieved_user = User.get_by_username('testuser')
assert retrieved_user is not None
assert retrieved_user.email == '[email protected]'
assert retrieved_user.check_password('password123')
def test_post_crud(app):
with app.app_context():
# Create test user
user = User(username='author', email='[email protected]')
user.set_password('pass')
user.save()
# Create post
post = Post(title='Test Post', content='Test content', author_id=user.id)
post.save()
# Read post
retrieved_post = Post.get_by_id(post.id)
assert retrieved_post.title == 'Test Post'
assert retrieved_post.author.username == 'author' # Test relationship
# Update post
retrieved_post.title = 'Updated Title'
retrieved_post.save()
updated_post = Post.get_by_id(post.id)
assert updated_post.title == 'Updated Title'
# Delete post
retrieved_post.delete()
assert Post.get_by_id(post.id) is None
def test_pagination(app):
with app.app_context():
# Create test user
user = User(username='testauthor', email='[email protected]')
user.set_password('password')
user.save()
# Create multiple posts
for i in range(15):
post = Post(title=f'Post {i}', content=f'Content {i}', author_id=user.id)
post.save()
# Test pagination
pagination = Post.get_all(page=1, per_page=10)
assert len(pagination.items) == 10
assert pagination.total == 15
assert pagination.pages == 2
# Test second page
pagination2 = Post.get_all(page=2, per_page=10)
assert len(pagination2.items) == 5
Jalankan test dengan:
pytest
Caching untuk Performance
Gunakan Flask-Caching untuk improve performance:
from flask_caching import Cache
cache = Cache()
# In __init__.py
cache.init_app(app, config={'CACHE_TYPE': 'simple'})
# In routes
@main.route('/posts')
@cache.cached(timeout=300) # Cache for 5 minutes
def posts():
posts = Post.get_all()
return render_template('posts.html', posts=posts)
Background Tasks dengan Celery
Untuk tasks yang berat, gunakan Celery:
# tasks.py
from celery import Celery
from app import create_app
celery = Celery('tasks', broker='redis://localhost:6379/0')
@celery.task
def send_email_notification(user_email, post_title):
# Send email logic here
pass
# In routes
from .tasks import send_email_notification
@main.route('/post/new', methods=['POST'])
def new_post():
# ... post creation logic ...
send_email_notification.delay(user.email, post.title)
# ...
API Documentation dengan Swagger
Gunakan Flask-RESTX untuk API documentation:
from flask_restx import Api, Resource, fields
api = Api(app, version='1.0', title='Flask CRUD API', description='A simple CRUD API')
post_model = api.model('Post', {
'title': fields.String(required=True, description='Post title'),
'content': fields.String(required=True, description='Post content'),
})
@api.route('/api/posts')
class PostsResource(Resource):
@api.marshal_list_with(post_model)
def get(self):
posts = Post.get_all()
return posts
@api.expect(post_model)
def post(self):
data = api.payload
post = Post(title=data['title'], content=data['content'], author_id=1)
post.save()
return {'message': 'Post created'}, 201
Deployment dengan Gunicorn
Untuk production deployment, gunakan Gunicorn:
pip install gunicorn
# Run with gunicorn
gunicorn -w 4 -b 0.0.0.0:8000 run:app
# Or create systemd service
sudo nano /etc/systemd/system/flask-app.service
Isi service file:
[Unit]
Description=Flask CRUD App
After=network.target
[Service]
User=www-data
Group=www-data
WorkingDirectory=/path/to/flask-crud-app
Environment="PATH=/path/to/flask-crud-app/venv/bin"
ExecStart=/path/to/flask-crud-app/venv/bin/gunicorn -w 4 -b 127.0.0.1:8000 run:app
Restart=always
[Install]
WantedBy=multi-user.target
Kesimpulan
Kita telah berhasil membuat aplikasi web CRUD lengkap dengan Flask dan MySQL menggunakan SQLAlchemy ORM! Dari basic setup sampai fitur-fitur advanced seperti authentication, testing, caching, dan deployment.
Key takeaways:
- Flask memberikan fleksibilitas tinggi untuk development
- SQLAlchemy ORM membuat interaksi database lebih type-safe dan maintainable
- Database migration dengan Flask-Migrate mempermudah schema changes
- Virtual environment penting untuk dependency management
- Struktur folder yang baik memudahkan maintenance
- Testing harus menjadi bagian dari development workflow
- Caching dan background tasks meningkatkan performance
- Documentation API membantu integration dengan frontend
Kamu bisa extend aplikasi ini dengan fitur-fitur lain seperti file upload, real-time notifications, atau integration dengan external APIs. Selamat mencoba dan happy coding!
Jika ada pertanyaan, jangan ragu untuk diskusi di komentar atau forum komunitas.