Building a Simple CRUD Web Application with Flask and MySQL

Introduction: Flask for Beginners

Hello fellow developers! In this article, we’ll learn together how to build a simple CRUD (Create, Read, Update, Delete) web application using the Flask framework and MySQL database. Flask is one of the most popular Python web frameworks due to its simplicity while remaining powerful.

We’ll start from scratch: environment setup, installing dependencies, creating database, implementing CRUD operations, up to advanced features like authentication, testing, and deployment. Perfect for those just starting with Flask or wanting to refresh their knowledge about web development with Python.

Table of Contents

• Why Flask + MySQL?
• Environment Preparation
• MySQL Database Setup
• Basic CRUD Implementation
• Good Folder Structure
• Authentication and Authorization
• Error Handling and Validation
• Testing with pytest
• Caching for Performance
• Background Tasks with Celery
• API Documentation with Swagger
• Deployment with Gunicorn
• Conclusion

Why Flask + MySQL?

Flask is a lightweight yet extensible micro-framework for Python web development. Unlike Django which comes with “batteries included”, Flask gives you complete freedom to choose the tools you need. This makes Flask very suitable for:

• Microservices architecture: Lightweight and easy to scale
• API development: Easy integration with modern frontends
• Prototyping: Fast development without boilerplate code
• Learning purposes: Gentle learning curve for beginners

MySQL as the database choice because:

• Reliable: Proven in production environments
• Fast: Excellent query performance
• Widely used: Many tools and community support
• ACID compliant: Data integrity guaranteed

Environment Preparation

Install Python

Make sure Python 3.8+ is installed on your system. Check with:

python --version
# or
python3 --version

If not available, download from python.org or use your system’s package manager.

Virtual Environment

Always use virtual environment for each Python project. This prevents dependency conflicts between projects.

# Create project folder
mkdir flask-crud-app
cd flask-crud-app

# Create virtual environment
python -m venv venv

# Activate virtual environment
# Windows:
venv\Scripts\activate
# Linux/Mac:
source venv/bin/activate

Install Dependencies

Install Flask and supporting libraries:

pip install Flask flask-sqlalchemy pymysql python-dotenv pytest flask-testing

We’ll install additional libraries as development progresses. For SQLAlchemy with MySQL, we use PyMySQL as the driver.

MySQL Database Setup

Install MySQL Server

# Ubuntu/Debian
sudo apt update
sudo apt install mysql-server

# CentOS/RHEL
sudo yum install mysql-server

# macOS with Homebrew
brew install mysql

Start MySQL service and setup:

sudo systemctl start mysql  # Linux
sudo systemctl enable mysql

# Setup secure installation
sudo mysql_secure_installation

Create Database and Tables

Login to MySQL as root:

mysql -u root -p

Create database and tables for our CRUD example:

-- Create database
CREATE DATABASE flask_crud_app;

-- Use database
USE flask_crud_app;

-- Create users table (for authentication later)
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
);

-- Create posts table (for 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');

Basic CRUD Implementation

Application Structure

Create folder structure like this:

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

Configuration File

Create 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 to 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

Create 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()

Benefits of using SQLAlchemy:

• Type-safe: Column types are clearly defined
• Relationships: Foreign keys and relationships are built-in
• Query builder: More readable and maintainable
• Migration support: Easy database schema changes
• Protection: Built-in SQL injection protection
• Pagination: Built-in pagination support

Initialize Database

Create 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

CRUD Routes

Create 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'))

Good Folder Structure

Good folder organization helps with application maintainability:

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 with Flask-Migrate

When using SQLAlchemy, you need a tool to manage database schema changes. Flask-Migrate (wrapper for Alembic) is perfect for this.

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

Migration Workflow

Every time you change a model:

# 1. Edit model in models.py
# 2. Generate migration script
flask db migrate -m "Add column email_verified to users"

# 3. Review migration file in migrations/versions/
# 4. Apply migration
flask db upgrade

To rollback:

# Rollback one version
flask db downgrade

# Rollback to specific version
flask db downgrade <revision_id>

Migration Tips

• Always review migration scripts before applying
• Backup database before running migrations in production
• Test migrations in development environment first
• Commit migration files to version control

Authentication and Authorization

Implement login/logout with sessions:

# 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 and Validation

Create custom error handlers and 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 and Indexes

Add constraints and indexes for data integrity and 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 for 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

Use lazy loading and eager loading appropriately:

# Lazy loading (default) - separate queries
post = Post.query.get(1)
print(post.author.username)  # Triggers second query

# Eager loading with joinedload - one query with JOIN
from sqlalchemy.orm import joinedload

posts = Post.query.options(joinedload(Post.author)).all()
for post in posts:
    print(post.author.username)  # No additional query

# Subquery loading - two queries but more efficient for one-to-many
from sqlalchemy.orm import subqueryload

users = User.query.options(subqueryload(User.posts)).all()

# Select specific columns to reduce memory
posts = db.session.query(Post.id, Post.title, Post.created_at).all()

Custom Query Methods

Add helper methods for frequently used queries:

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

Create 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 with pytest

Create 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

Run tests with:

pytest

Caching for Performance

Use Flask-Caching to 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 with Celery

For heavy tasks, use 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 with Swagger

Use Flask-RESTX for 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 with Gunicorn

For production deployment, use 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

Service file content:

[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

Conclusion

We have successfully created a complete CRUD web application with Flask and MySQL using SQLAlchemy ORM! From basic setup to advanced features like authentication, testing, caching, and deployment.

Key takeaways:

• Flask provides high flexibility for development
• SQLAlchemy ORM makes database interactions type-safe and maintainable
• Database migrations with Flask-Migrate simplify schema changes
• Virtual environment is crucial for dependency management
• Good folder structure makes maintenance easier
• Testing should be part of the development workflow
• Caching and background tasks improve performance
• API documentation helps with frontend integration

You can extend this application with other features like file upload, real-time notifications, or integration with external APIs. Happy coding!

If you have any questions, feel free to discuss in the comments or community forums.